Compare commits

...

11 Commits

Author SHA1 Message Date
680ac6e980 Merge pull request 'feat: add detailed employee publications storage and MCP docs' (#27) from feature/employee-publications-db into main
Reviewed-on: #27
2026-05-15 14:40:29 +00:00
Anton
dbaf3af468 feat: add detailed employee publications storage and MCP docs 2026-05-15 17:39:41 +03:00
2819a6c334 Merge pull request 'fix: add runtime schema guard for skipped count' (#26) from fix/runtime-schema-skipped-count into main
Reviewed-on: #26
2026-05-14 10:30:06 +00:00
Anton
41fb54c5e7 fix: add runtime schema guard for skipped count 2026-05-14 13:29:27 +03:00
4b91effee3 Merge pull request 'feat: adds crawl resource cache' (#25) from feature/crawl-resource-cache into main
Reviewed-on: #25
2026-05-14 09:27:06 +00:00
Anton
6724b3f369 feat: adds crawl resource cache 2026-05-14 12:21:44 +03:00
1791ad8d4d Merge pull request 'chore: adds additional mcp-description file to gitignore' (#24) from docs/mcp-description into main
Reviewed-on: #24
2026-05-14 08:52:29 +00:00
Anton
993888b003 chore: adds additional mcp-description file to gitignore 2026-05-14 11:51:51 +03:00
5180b89b81 Merge pull request 'feat: add dataset checkpoint sync for MCP' (#23) from feature/dataset-version-sync into main
Reviewed-on: #23
2026-05-14 08:01:26 +00:00
Anton
29451ccee1 feat: add dataset checkpoint sync for MCP 2026-05-14 11:00:46 +03:00
a3ff9c6e9c Merge pull request 'fix: separate news from publications and add employee refresh' (#22) from fix/publications-news-refresh into main
Reviewed-on: #22
2026-05-13 13:12:06 +00:00
28 changed files with 2446 additions and 87 deletions

1
.gitignore vendored
View File

@@ -8,3 +8,4 @@ pytest-cache-files-*/
.coverage
htmlcov/
postgres_data/
MCP_DESCRIPTION.md

648
MCP_DESCRIPTION.md Normal file
View File

@@ -0,0 +1,648 @@
# MCP: описание работы, структуры и тулзов
Документ описывает MCP endpoint сервиса `miem-employees` по текущей реализации в `app/mcp.py`.
## Где находится MCP
- FastAPI router: `app.mcp.router`
- Подключение к приложению: `app/main.py`
- HTTP endpoint: `POST /mcp`
- Локально при обычном запуске API: `http://localhost:8000/mcp`
- В Docker Compose через отдельный сервис `mcp`: `http://localhost:8001/mcp`
- Авторизация на уровне приложения: отсутствует. Заголовок `Authorization` не проверяется и не влияет на ответ.
Если доступ к MCP нужно ограничить, это должно делаться внешним контуром: bind на localhost, VPN, firewall, reverse proxy или отдельная сетевая политика.
## Протокол
Endpoint принимает JSON-RPC 2.0 over HTTP.
Общий формат запроса:
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}
```
Общий формат успешного ответа:
```json
{
"jsonrpc": "2.0",
"id": 1,
"result": {}
}
```
Общий формат ошибки:
```json
{
"jsonrpc": "2.0",
"id": 1,
"error": {
"code": -32601,
"message": "Method not found"
}
}
```
Поддерживаемая версия MCP-протокола:
```text
2024-11-05
```
Имя сервиса:
```text
miem-employees
```
Версия сервера берется из `app.version.BACKEND_VERSION`.
## Поддерживаемые JSON-RPC методы
### initialize
Возвращает метаданные MCP-сервера и capabilities.
Запрос:
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {}
}
```
Ответ:
```json
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"protocolVersion": "2024-11-05",
"serverInfo": {
"name": "miem-employees",
"version": "0.5.0"
},
"capabilities": {
"tools": {}
}
}
}
```
### tools/list
Возвращает список доступных tools с JSON Schema для аргументов.
Запрос:
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}
```
Ответ содержит массив `result.tools`.
### tools/call
Вызывает один tool по имени.
Запрос:
```json
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "search_employees",
"arguments": {
"query": "Сергеев",
"limit": 20
}
}
}
```
Ответ tool всегда заворачивается в MCP content-массив:
```json
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"content": [
{
"type": "text",
"text": "{\"items\":[]}"
}
]
}
}
```
Поле `text` содержит сериализованный JSON с `ensure_ascii=false`. Клиент должен распарсить это поле как JSON, если ему нужна структурированная нагрузка.
## Ошибки
- Неизвестный JSON-RPC метод: `code = -32601`, `message = "Method not found"`.
- Исключения при обработке tool: `code = -32000`, `message` содержит текст исключения.
- Если сущность не найдена внутри отдельных tools, HTTP и JSON-RPC ответ остаются успешными, а полезная нагрузка содержит `{"error": "not_found"}`.
## Источники данных
MCP читает данные из основной базы через SQLAlchemy session из `app.db.get_db`.
Основные таблицы и модели:
- `employees`: текущая карточка сотрудника, статус, профиль, `current_data`, checksum.
- `employee_publications`: нормализованные публикации сотрудников с авторами, DOI, аннотацией, описанием, citation text и raw JSON из HSE Publications.
- `crawl_runs`: история запусков парсинга.
- `crawl_run_employee_changes`: детальные изменения сотрудников в рамках запуска.
- `crawl_errors`: ошибки парсинга в рамках запуска.
- `dataset_versions`: версии полного набора сотрудников.
- `dataset_version_items`: состав конкретной версии набора сотрудников.
## Общая структура employee payload
Краткая карточка сотрудника:
```json
{
"profile_key": "staff:avsergeev",
"profile_id": "avsergeev",
"full_name": "Сергеев Алексей Викторович",
"status": "active",
"canonical_url": "https://www.hse.ru/staff/avsergeev",
"last_seen_at": "2026-05-14T10:00:00+00:00",
"dismissed_at": null
}
```
В sync payload дополнительно отдается `checksum`.
Полная карточка дополнительно содержит:
```json
{
"data": {
"contacts": {},
"sections": []
}
}
```
`data` соответствует распарсенному JSON профиля сотрудника. Внутри `sections` могут быть секции с публикациями, курсами, ВКР, таблицами, ссылками и произвольными текстовыми блоками.
## Tools
### get_service_info
Назначение: вернуть метаданные сервиса, список tools и текущую версию набора сотрудников.
Аргументы: отсутствуют.
Возвращает:
```json
{
"service_name": "miem-employees",
"backend_version": "0.5.0",
"protocolVersion": "2024-11-05",
"tools": [],
"dataset": {
"hash": "sha256",
"previous_hash": "sha256 или null",
"created_at": "2026-05-14T10:00:00+00:00",
"crawl_run_id": 123,
"employee_count": 100,
"active_count": 95,
"dismissed_count": 5
}
}
```
Особенность: перед ответом сервис создает актуальную `dataset_version`, если текущий набор сотрудников еще не имеет версии.
### sync_employees
Назначение: синхронизировать клиентский кэш сотрудников по hash набора данных.
Аргументы:
```json
{
"client_hash": "sha256 или null",
"include_data": true
}
```
- `client_hash`: hash версии, которая уже есть у клиента. Если не передан, отдается полный snapshot.
- `include_data`: управляет включением полного `data` в карточки сотрудников. По умолчанию `true`.
Полный ответ без `client_hash`:
```json
{
"mode": "full",
"from_hash": null,
"to_hash": "current-sha256",
"dataset": {},
"items": []
}
```
Если клиентский hash совпадает с текущим:
```json
{
"mode": "delta",
"from_hash": "current-sha256",
"to_hash": "current-sha256",
"dataset": {},
"changes": {
"added": [],
"updated": [],
"dismissed": [],
"removed": []
}
}
```
Если `client_hash` неизвестен серверу:
```json
{
"mode": "full",
"from_hash": "missing",
"to_hash": "current-sha256",
"dataset": {},
"items": [],
"reason": "unknown_client_hash"
}
```
Если `client_hash` найден и отличается от текущего:
```json
{
"mode": "delta",
"from_hash": "old-sha256",
"to_hash": "current-sha256",
"dataset": {},
"changes": {
"added": [],
"updated": [],
"dismissed": [],
"removed": []
}
}
```
Логика delta:
- `added`: сотрудник появился в новой версии.
- `updated`: изменился checksum или статус, и сотрудник активен.
- `dismissed`: сотрудник есть в новой версии, но получил статус `dismissed`.
- `removed`: `profile_key` был в старой версии, но отсутствует в новой.
Hash набора считается по отсортированному списку `{profile_key, status, checksum}`.
### search_employees
Назначение: найти сотрудников по ФИО или canonical URL.
Аргументы:
```json
{
"query": "Сергеев",
"status": "active",
"limit": 20
}
```
- `query`: обязательный по schema, но в коде пустая строка означает поиск без текстового фильтра.
- `status`: опционально, только `active` или `dismissed`.
- `limit`: максимум 100, по умолчанию 20.
Возвращает массив кратких employee payload без `data`:
```json
[
{
"profile_key": "staff:avsergeev",
"profile_id": "avsergeev",
"full_name": "Сергеев Алексей Викторович",
"status": "active",
"canonical_url": "https://www.hse.ru/staff/avsergeev",
"last_seen_at": "2026-05-14T10:00:00+00:00",
"dismissed_at": null
}
]
```
### get_employee
Назначение: получить одну карточку сотрудника.
Аргументы:
```json
{
"profile_id_or_url": "avsergeev"
}
```
Поиск выполняется по:
- `profile_key`
- `profile_id`
- точному `canonical_url`
- частичному совпадению `canonical_url`
Возвращает полный employee payload с `data`.
Если сотрудник не найден:
```json
{
"error": "not_found"
}
```
### list_employee_publications
Назначение: вернуть публикации сотрудника. Если есть нормализованные строки в `employee_publications`, tool возвращает детальные публикационные данные: авторов, DOI, аннотацию, описание, citation text, год, тип, язык, статус и ссылки. Если детальная таблица еще не заполнена, tool использует старый fallback из `employees.current_data.sections[].publications`.
Аргументы:
```json
{
"profile_id_or_url": "avsergeev"
}
```
Поиск сотрудника выполняется так же, как в `get_employee`: по `profile_key`, `profile_id`, точному или частичному `canonical_url`.
Порядок источников:
- сначала `employee_publications`, отсортированные по году, названию и внутреннему id;
- если записей нет, секции `current_data.sections` с `type = "publications"` и массивами `publications`.
Ответ:
```json
{
"employee": {
"profile_key": "org_person:803294906",
"profile_id": "803294906",
"full_name": "Борисов Сергей Петрович",
"status": "active",
"canonical_url": "https://www.hse.ru/org/persons/803294906",
"last_seen_at": "2026-05-14T10:00:00+00:00",
"dismissed_at": null
},
"items": [
{
"id": "888959076",
"publication_id": "888959076",
"title": "Название публикации",
"text": "Краткое описание или citation",
"url": "https://publications.hse.ru/view/888959076",
"year": 2023,
"type": "ARTICLE",
"publication_type": "ARTICLE",
"language": "ru",
"status": 1,
"doi_url": "https://doi.org/10.53921/18195822_2023_23_4_624",
"other_url": "https://example.test",
"document_url": "https://example.test/file.pdf",
"citation_text": "Авторы. Название публикации // Журнал. 2023.",
"annotation": {
"ru": "Аннотация",
"en": "Abstract"
},
"description": {
"main": "Авторы. Название публикации // Журнал. 2023."
},
"authors": [
{
"id": "803294906",
"href": "https://www.hse.ru/org/persons/803294906",
"title_ru": "Борисов С. П.",
"title_en": "",
"reverse_title_ru": "С. П. Борисов",
"reverse_title_en": "",
"alt_name": "S. P. Borisov",
"other_name": null,
"is_current_employee": true
}
]
}
]
}
```
В fallback-режиме из `current_data` старые элементы могут содержать только базовые поля `title`, `text`, `url` и `id`.
Если сотрудник не найден:
```json
{
"items": []
}
```
Если сотрудник найден, но публикаций нет:
```json
{
"employee": {},
"items": []
}
```
### list_employee_courses
Назначение: вернуть курсы преподавания сотрудника из распарсенных секций профиля.
Аргументы:
```json
{
"profile_id_or_url": "avsergeev"
}
```
Сервис ищет секции `current_data.sections` с `type = "courses_by_year"` и объединяет массивы `courses`.
Ответ:
```json
{
"employee": {},
"items": [
{
"title": "Название курса",
"url": "https://..."
}
]
}
```
Если сотрудник или данные профиля отсутствуют:
```json
{
"items": []
}
```
### get_crawl_status
Назначение: вернуть последний запуск парсинга.
Аргументы: отсутствуют.
Ответ:
```json
{
"id": 123,
"status": "completed",
"source_url": "https://miem.hse.ru/persons",
"started_at": "2026-05-14T10:00:00+00:00",
"finished_at": "2026-05-14T10:10:00+00:00",
"found_count": 100,
"parsed_count": 98,
"error_count": 2,
"dismissed_count": 1
}
```
Если запусков еще не было:
```json
{
"status": "never_run"
}
```
### get_crawl_run_details
Назначение: вернуть детальную информацию по конкретному запуску парсинга: summary, изменения сотрудников и ошибки.
Аргументы:
```json
{
"run_id": 123
}
```
Ответ:
```json
{
"id": 123,
"source_url": "https://miem.hse.ru/persons",
"status": "completed",
"status_display": "Завершен",
"started_at": "2026-05-14T10:00:00+00:00",
"finished_at": "2026-05-14T10:10:00+00:00",
"started_display": "14.05.2026 13:00",
"finished_display": "14.05.2026 13:10",
"found_count": 100,
"parsed_count": 98,
"new_count": 3,
"error_count": 2,
"dismissed_count": 1,
"processed_count": 100,
"progress_percent": 100.0,
"message": null,
"changes_detail_available": true,
"changes": {
"new": [],
"missing_from_source": [],
"dismissed": []
},
"errors": []
}
```
Если запуск не найден:
```json
{
"error": "not_found"
}
```
## Примеры curl
Список tools:
```bash
curl http://localhost:8001/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'
```
Поиск сотрудника:
```bash
curl http://localhost:8001/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"search_employees","arguments":{"query":"Сергеев","limit":5}}}'
```
Полная синхронизация:
```bash
curl http://localhost:8001/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"sync_employees","arguments":{"include_data":false}}}'
```
Delta-синхронизация:
```bash
curl http://localhost:8001/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"sync_employees","arguments":{"client_hash":"known-sha256","include_data":true}}}'
```
## Как MCP используется клиентом
1. Клиент вызывает `initialize` и проверяет `protocolVersion`.
2. Клиент вызывает `tools/list`, чтобы получить актуальный список tools и input schemas.
3. Для поиска и точечных запросов клиент вызывает `tools/call` с `search_employees`, `get_employee`, `list_employee_publications`, `list_employee_courses`, `get_crawl_status` или `get_crawl_run_details`.
4. Для локального кэша клиент вызывает `get_service_info` или `sync_employees`.
5. Клиент хранит последний `dataset.hash`.
6. При следующей синхронизации клиент передает hash как `client_hash`.
7. Сервер возвращает пустую delta, delta с изменениями или полный snapshot, если hash неизвестен.
## Важные особенности реализации
- MCP endpoint read-only: tools не запускают парсинг и не меняют сотрудников напрямую.
- `get_service_info` и `sync_employees` могут создать новую запись `dataset_versions`, если состояние сотрудников изменилось и новой версии еще нет.
- Все tool payloads возвращаются как JSON-строка внутри `content[0].text`.
- `search_employees` ищет через `ilike` по `full_name` и `canonical_url`.
- `get_employee` допускает частичный URL, поэтому строка `133709486` может найти `https://www.hse.ru/org/persons/133709486`.
- Временные значения сериализуются через `isoformat()`, display-поля для админских payload формируются в часовом поясе `Europe/Moscow`.

View File

@@ -9,7 +9,7 @@
- `mcp`: открытый HTTP MCP endpoint для ИИ-агентов.
- `postgres`: основная БД.
Парсер использует фиксированный источник сотрудников, по умолчанию `https://miem.hse.ru/persons`. Для каждой карточки сохраняются ФИО, должности, год начала работы, контакты, идентификаторы, вкладки профиля, секции, публикации, курсы, ВКР, JSON-снапшот и сжатый HTML-снапшот. Ссылки обходятся только из меню профиля самого сотрудника (`person-menu`), например `#sci`, `#teaching`, `#main`.
Парсер использует фиксированный источник сотрудников, по умолчанию `https://miem.hse.ru/persons`. Для каждой карточки сохраняются ФИО, должности, год начала работы, контакты, идентификаторы, вкладки профиля, секции, публикации, курсы, ВКР, JSON-снапшот и сжатый HTML-снапшот. Детальные публикации дополнительно нормализуются в отдельную таблицу `employee_publications`. Ссылки обходятся только из меню профиля самого сотрудника (`person-menu`), например `#sci`, `#teaching`, `#main`.
## Переменные окружения
@@ -44,7 +44,6 @@ uvicorn app.main:app --reload
- `Dashboard`: общая статистика, последний добавленный сотрудник, прогресс текущего/последнего парсинга и ручной запуск.
- `Directory`: настраиваемая таблица сотрудников с фильтрами, сортировкой, пагинацией и выбором колонок.
- `Employees`: простая legacy-таблица сотрудников.
- `Runs`: история запусков, ошибки и progress bar.
## Docker Compose
@@ -59,7 +58,20 @@ docker compose up --build
- MCP: `http://localhost:8001/mcp`
- Postgres: `localhost:5432`
Таблицы создаются приложением при старте. SQL-миграция для ручного применения лежит в `migrations/001_init.sql`.
Таблицы создаются приложением при старте. При обновлении существующей базы приложение также добавляет недостающие runtime-колонки, например `crawl_runs.skipped_count`. SQL-миграции для ручного применения лежат в `migrations/`.
## Наполнение БД
Основная карточка сотрудника хранится в `employees`: профиль, статус, даты обнаружения/увольнения, текущий JSON `current_data`, checksum и версия парсера. История успешных изменений сохраняется в `employee_snapshots` вместе с JSON-снимком и сжатым HTML профиля.
Публикации теперь хранятся в двух видах:
- краткий список остается внутри `employees.current_data.sections[].publications` для обратной совместимости;
- детальные записи сохраняются в `employee_publications` и связываются с сотрудником через `employee_id`.
`employee_publications` содержит `publication_id`, название, год, тип публикации, язык, статус, ссылку на карточку HSE Publications, DOI, внешние/document-ссылки, citation text, аннотацию, описание, авторов, raw JSON ответа `searchPubs` и `source_hash` для безопасного повторного upsert. Уникальность поддерживается по `(employee_id, publication_id)` и `(employee_id, source_hash)`, поэтому повторный crawl не должен создавать дубликаты.
`list_employee_publications` сначала читает `employee_publications`; если детальных строк еще нет, возвращает старые публикации из `current_data`.
## Парсинг
@@ -74,10 +86,12 @@ curl -X POST http://localhost:8000/api/crawl-runs --cookie "miem_admin_session=.
- найденные сотрудники получают статус `active` и обновленный `last_seen_at`;
- новые сотрудники добавляются в `employees`;
- количество новых сотрудников за запуск сохраняется в `crawl_runs.new_count`;
- публикации из HSE Publications записываются в `employee_publications`, а краткий список остается в JSON профиля;
- активные сотрудники, исчезнувшие из текущего списка источника, получают статус `dismissed` и `dismissed_at`;
- каждый успешный разбор сохраняет запись в `employee_snapshots`.
- каждый успешный новый или измененный разбор сохраняет запись в `employee_snapshots`;
- неизмененные профили учитываются в `crawl_runs.skipped_count` и не получают новый snapshot.
Во время выполнения парсинга `found_count`, `parsed_count` и `error_count` обновляются в базе. Админка опрашивает `/api/crawl-runs/latest` и показывает прогресс как `parsed_count + error_count / found_count`.
Во время выполнения парсинга `found_count`, `parsed_count`, `skipped_count` и `error_count` обновляются в базе. Админка опрашивает `/api/crawl-runs/latest` и показывает прогресс как `(parsed_count + skipped_count + error_count) / found_count`.
## MCP
@@ -85,13 +99,18 @@ Endpoint: `POST /mcp`, без авторизации на уровне прил
Поддерживаемые tools:
- `get_service_info()`
- `sync_employees(client_hash?, include_data?)`
- `search_employees(query, status?, limit?)`
- `get_employee(profile_id_or_url)`
- `list_employee_publications(profile_id_or_url)`
- `list_employee_publications(profile_id_or_url)` — публикации сотрудника; при наличии данных из `employee_publications` возвращает авторов, DOI, аннотацию, описание, citation text, год, тип, язык, статус и ссылку HSE Publications.
- `list_employee_courses(profile_id_or_url)`
- `get_crawl_status()`
- `get_crawl_run_details(run_id)`
Пример локального legacy-режима со статическим токеном:
`get_service_info` возвращает метаданные сервиса, список tools и текущую версию набора сотрудников. `sync_employees` отдает полный snapshot или delta по `client_hash`; checksum набора строится по сотрудникам, их статусам и текущим checksums. Ответы tools возвращаются как JSON-строка внутри MCP `content[0].text`.
Пример локального запроса списка tools:
```bash
curl http://localhost:8001/mcp \
@@ -110,4 +129,4 @@ docker compose exec postgres pg_dump -U miem miem_workers > backup.sql
docker compose down
```
Версия сервиса: `0.4.5`. Админка всегда показывает версии backend и frontend в footer.
Версия сервиса: `0.6.2`. Админка всегда показывает версии backend и frontend в footer.

View File

@@ -1,6 +1,6 @@
from collections.abc import Generator
from sqlalchemy import create_engine
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
from app.config import get_settings
@@ -25,6 +25,24 @@ def init_db() -> None:
import app.models # noqa: F401
Base.metadata.create_all(bind=engine)
_ensure_runtime_schema()
def _ensure_runtime_schema() -> None:
import app.models as models
inspector = inspect(engine)
table_names = set(inspector.get_table_names())
if "employees" in table_names and "employee_publications" not in table_names:
models.EmployeePublication.__table__.create(bind=engine, checkfirst=True)
inspector = inspect(engine)
table_names = set(inspector.get_table_names())
if "crawl_runs" not in table_names:
return
crawl_run_columns = {column["name"] for column in inspector.get_columns("crawl_runs")}
if "skipped_count" not in crawl_run_columns:
with engine.begin() as connection:
connection.execute(text("ALTER TABLE crawl_runs ADD COLUMN skipped_count INTEGER NOT NULL DEFAULT 0"))
def get_db() -> Generator[Session, None, None]:

View File

@@ -5,14 +5,33 @@ from sqlalchemy import desc, or_, select
from sqlalchemy.orm import Session
from app.db import get_db
from app.models import CrawlRun, Employee
from app.models import CrawlRun, Employee, EmployeePublication
from app.services.admin_data import run_detail_payload
from app.services.dataset_versions import service_info_payload, sync_employees_payload
from app.version import BACKEND_VERSION
router = APIRouter(prefix="/mcp")
PROTOCOL_VERSION = "2024-11-05"
SERVICE_NAME = "miem-employees"
TOOLS = [
{
"name": "get_service_info",
"description": "Return service metadata, supported tools, and current dataset version.",
"inputSchema": {"type": "object", "properties": {}},
},
{
"name": "sync_employees",
"description": "Synchronize employees by dataset hash. Returns a full snapshot or a delta from client_hash.",
"inputSchema": {
"type": "object",
"properties": {
"client_hash": {"type": "string"},
"include_data": {"type": "boolean", "default": True},
},
},
},
{
"name": "search_employees",
"description": "Search MIEM employees by name or profile URL.",
@@ -33,7 +52,10 @@ TOOLS = [
},
{
"name": "list_employee_publications",
"description": "List publications parsed from an employee profile.",
"description": (
"List employee publications with detailed fields when available: authors, DOI URL, annotation, "
"description, citation text, year, publication type, language, status, and HSE Publications URL."
),
"inputSchema": {"type": "object", "properties": {"profile_id_or_url": {"type": "string"}}, "required": ["profile_id_or_url"]},
},
{
@@ -71,8 +93,8 @@ async def mcp_http(
try:
if method == "initialize":
result = {
"protocolVersion": "2024-11-05",
"serverInfo": {"name": "miem-employees", "version": BACKEND_VERSION},
"protocolVersion": PROTOCOL_VERSION,
"serverInfo": {"name": SERVICE_NAME, "version": BACKEND_VERSION},
"capabilities": {"tools": {}},
}
elif method == "tools/list":
@@ -87,6 +109,24 @@ async def mcp_http(
def _call_tool(db: Session, name: str, arguments: dict) -> dict:
if name == "get_service_info":
return _tool_response(
service_info_payload(
db,
tools=TOOLS,
service_name=SERVICE_NAME,
backend_version=BACKEND_VERSION,
protocol_version=PROTOCOL_VERSION,
)
)
if name == "sync_employees":
return _tool_response(
sync_employees_payload(
db,
client_hash=arguments.get("client_hash"),
include_data=bool(arguments.get("include_data", True)),
)
)
if name == "search_employees":
return _tool_response(_search_employees(db, arguments))
if name == "get_employee":
@@ -134,8 +174,14 @@ def _find_employee(db: Session, value: str) -> Employee | None:
def _collect_section_items(employee: Employee | None, section_type: str) -> dict:
if not employee or not employee.current_data:
if not employee:
return {"items": []}
if section_type == "publications":
publications = _stored_publications(employee)
if publications:
return {"employee": _employee_payload(employee, include_data=False), "items": publications}
if not employee.current_data:
return {"employee": _employee_payload(employee, include_data=False), "items": []}
items = []
for section in employee.current_data.get("sections") or []:
if section.get("type") != section_type:
@@ -147,6 +193,41 @@ def _collect_section_items(employee: Employee | None, section_type: str) -> dict
return {"employee": _employee_payload(employee, include_data=False), "items": items}
def _stored_publications(employee: Employee) -> list[dict]:
return [_publication_payload(publication) for publication in sorted(employee.publications, key=_publication_sort_key)]
def _publication_sort_key(publication: EmployeePublication) -> tuple:
return (publication.year or 0, publication.title or "", publication.id)
def _publication_payload(publication: EmployeePublication) -> dict:
text = publication.citation_text or publication.title
payload = {
"id": publication.publication_id,
"publication_id": publication.publication_id,
"title": publication.title,
"text": text,
"url": publication.url,
}
optional = {
"year": publication.year,
"type": publication.publication_type,
"publication_type": publication.publication_type,
"language": publication.language,
"status": publication.status,
"doi_url": publication.doi_url,
"other_url": publication.other_url,
"document_url": publication.document_url,
"citation_text": publication.citation_text,
"annotation": publication.annotation,
"description": publication.description,
"authors": publication.authors,
}
payload.update({key: value for key, value in optional.items() if value not in (None, [], {})})
return payload
def _employee_payload(employee: Employee, include_data: bool = True) -> dict:
payload = {
"profile_key": employee.profile_key,
@@ -171,6 +252,7 @@ def _run_payload(run: CrawlRun) -> dict:
"finished_at": run.finished_at.isoformat() if run.finished_at else None,
"found_count": run.found_count,
"parsed_count": run.parsed_count,
"skipped_count": run.skipped_count,
"error_count": run.error_count,
"dismissed_count": run.dismissed_count,
}

View File

@@ -41,6 +41,7 @@ class Employee(Base):
snapshots: Mapped[list["EmployeeSnapshot"]] = relationship(back_populates="employee")
tabs: Mapped[list["ProfileTab"]] = relationship(back_populates="employee", cascade="all, delete-orphan")
publications: Mapped[list["EmployeePublication"]] = relationship(back_populates="employee", cascade="all, delete-orphan")
crawl_run_changes: Mapped[list["CrawlRunEmployeeChange"]] = relationship(back_populates="employee")
@@ -60,6 +61,42 @@ class EmployeeSnapshot(Base):
employee: Mapped[Employee] = relationship(back_populates="snapshots")
class EmployeePublication(Base):
__tablename__ = "employee_publications"
__table_args__ = (
UniqueConstraint("employee_id", "publication_id", name="uq_employee_publications_employee_publication"),
UniqueConstraint("employee_id", "source_hash", name="uq_employee_publications_employee_source_hash"),
Index("ix_employee_publications_employee_id", "employee_id"),
Index("ix_employee_publications_publication_id", "publication_id"),
Index("ix_employee_publications_doi_url", "doi_url"),
Index("ix_employee_publications_year", "year"),
Index("ix_employee_publications_publication_type", "publication_type"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
employee_id: Mapped[int] = mapped_column(ForeignKey("employees.id", ondelete="CASCADE"), nullable=False)
publication_id: Mapped[str | None] = mapped_column(String(64))
title: Mapped[str] = mapped_column(Text, nullable=False)
year: Mapped[int | None] = mapped_column(Integer)
publication_type: Mapped[str | None] = mapped_column(String(64))
language: Mapped[str | None] = mapped_column(String(16))
status: Mapped[int | None] = mapped_column(Integer)
url: Mapped[str | None] = mapped_column(Text)
doi_url: Mapped[str | None] = mapped_column(Text)
other_url: Mapped[str | None] = mapped_column(Text)
document_url: Mapped[str | None] = mapped_column(Text)
citation_text: Mapped[str | None] = mapped_column(Text)
annotation: Mapped[dict | None] = mapped_column(json_type)
description: Mapped[dict | None] = mapped_column(json_type)
authors: Mapped[list | None] = mapped_column(json_type)
raw_data: Mapped[dict | None] = mapped_column(json_type)
source_hash: Mapped[str] = mapped_column(String(64), nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, nullable=False)
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, onupdate=utcnow, nullable=False)
employee: Mapped[Employee] = relationship(back_populates="publications")
class CrawlRun(Base):
__tablename__ = "crawl_runs"
@@ -70,12 +107,14 @@ class CrawlRun(Base):
finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
found_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
parsed_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
skipped_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
new_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
error_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
dismissed_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
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):
@@ -134,3 +173,63 @@ class ParserSource(Base):
source_url: Mapped[str] = mapped_column(Text, nullable=False)
enabled: Mapped[bool] = mapped_column(default=True, nullable=False)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utcnow, nullable=False)
class ParseResourceCache(Base):
__tablename__ = "parse_resource_cache"
__table_args__ = (
UniqueConstraint("profile_key", "resource_key", "request_fingerprint", name="uq_parse_resource_cache_resource"),
Index("ix_parse_resource_cache_profile_key", "profile_key"),
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
profile_key: Mapped[str] = mapped_column(String(255), nullable=False)
resource_key: Mapped[str] = mapped_column(String(255), nullable=False)
method: Mapped[str] = mapped_column(String(16), nullable=False)
url: Mapped[str] = mapped_column(Text, nullable=False)
request_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False)
etag: Mapped[str | None] = mapped_column(Text)
last_modified: Mapped[str | None] = mapped_column(Text)
body_hash: Mapped[str] = mapped_column(String(64), nullable=False)
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()

View File

@@ -1,3 +1,5 @@
import hashlib
import json
import re
from urllib.parse import urljoin
@@ -149,22 +151,42 @@ def parse_person_profile(
headers: dict[str, str],
timeout: int,
use_playwright: bool = False,
resource_cache=None,
) -> dict | None:
normalized_url = normalize_profile_url(source_url)
if not normalized_url:
return None
response = session.get(normalized_url, headers=headers, timeout=timeout)
response.raise_for_status()
html = response.text
profile_type, profile_id = parse_profile_identity(normalized_url)
cache_profile_key = f"{profile_type}:{profile_id}"
resource_manifest = []
html = _fetch_text(
session,
normalized_url,
headers,
timeout,
resource_cache=resource_cache,
profile_key=cache_profile_key,
resource_key="main-html",
resource_manifest=resource_manifest,
)
if use_playwright:
html = _render_with_playwright(normalized_url, html)
soup = BeautifulSoup(html, "html.parser")
profile_type, profile_id = parse_profile_identity(normalized_url)
header = extract_person_header(soup, normalized_url)
tabs = extract_person_tabs(soup, normalized_url)
sections = extract_sections(soup, normalized_url)
sections = enrich_sections_from_hse_widgets(session, soup, normalized_url, headers, timeout, sections)
sections = enrich_sections_from_hse_widgets(
session,
soup,
normalized_url,
headers,
timeout,
sections,
resource_cache=resource_cache,
profile_key=cache_profile_key,
resource_manifest=resource_manifest,
)
internal_links = [tab["href"] for tab in tabs if tab.get("href")]
return {
@@ -181,6 +203,7 @@ def parse_person_profile(
"employee_internal_links": internal_links,
"parser_version": BACKEND_VERSION,
"_html": html,
"_resource_manifest": resource_manifest,
}
@@ -191,13 +214,33 @@ def enrich_sections_from_hse_widgets(
headers: dict[str, str],
timeout: int,
sections: list[dict],
resource_cache=None,
profile_key: str | None = None,
resource_manifest: list[dict] | None = None,
) -> list[dict]:
enriched = list(sections)
publications = _load_widget_publications(session, soup, headers, timeout)
publications = _load_widget_publications(
session,
soup,
headers,
timeout,
resource_cache=resource_cache,
profile_key=profile_key,
resource_manifest=resource_manifest,
)
if publications:
enriched = _upsert_publications_section(enriched, publications)
theses = _load_widget_graduation_theses(session, soup, source_url, headers, timeout)
theses = _load_widget_graduation_theses(
session,
soup,
source_url,
headers,
timeout,
resource_cache=resource_cache,
profile_key=profile_key,
resource_manifest=resource_manifest,
)
if theses:
enriched = _upsert_graduation_theses_section(enriched, theses)
return enriched
@@ -226,7 +269,16 @@ def _render_with_playwright(source_url: str, fallback_html: str) -> str:
return fallback_html
def _load_widget_publications(session: Session, soup: BeautifulSoup, headers: dict[str, str], timeout: int) -> list[dict]:
def _load_widget_publications(
session: Session,
soup: BeautifulSoup,
headers: dict[str, str],
timeout: int,
*,
resource_cache=None,
profile_key: str | None = None,
resource_manifest: list[dict] | None = None,
) -> list[dict]:
script = soup.select_one('script[data-widget-name="AuthorSearch"][data-author]')
if not script:
return []
@@ -251,6 +303,21 @@ def _load_widget_publications(session: Session, soup: BeautifulSoup, headers: di
},
}
try:
if resource_cache and profile_key:
text = _fetch_text(
session,
"https://publications.hse.ru/api/searchPubs",
headers,
timeout,
resource_cache=resource_cache,
profile_key=profile_key,
resource_key=f"publications-page-{page_id}",
resource_manifest=resource_manifest,
method="POST",
json_payload=payload,
)
data = json.loads(text)
else:
response = session.post(
"https://publications.hse.ru/api/searchPubs",
json=payload,
@@ -266,7 +333,7 @@ def _load_widget_publications(session: Session, soup: BeautifulSoup, headers: di
items = _extract_publication_items(result)
if not items:
break
publications.extend(_normalize_publication_item(item) for item in items)
publications.extend(_normalize_publication_item(item, author_id) for item in items)
total = int(result.get("total") or 0)
if not result.get("more") and len(publications) >= total:
@@ -309,6 +376,10 @@ def _load_widget_graduation_theses(
source_url: str,
headers: dict[str, str],
timeout: int,
*,
resource_cache=None,
profile_key: str | None = None,
resource_manifest: list[dict] | None = None,
) -> list[dict]:
script = soup.select_one('script[src*="/n/stat/vkr/app.js"][data-person-id]')
if not script:
@@ -320,9 +391,25 @@ def _load_widget_graduation_theses(
request_headers = {**headers, "x-portal-language": "ru"}
try:
url = urljoin(source_url, api_url)
params = {"supervisorId": person_id}
if resource_cache and profile_key:
text = _fetch_text(
session,
url,
request_headers,
timeout,
resource_cache=resource_cache,
profile_key=profile_key,
resource_key="graduation-theses",
resource_manifest=resource_manifest,
params=params,
)
data = json.loads(text)
else:
response = session.get(
urljoin(source_url, api_url),
params={"supervisorId": person_id},
url,
params=params,
headers=request_headers,
timeout=timeout,
)
@@ -488,20 +575,37 @@ def _parse_vkr_items(nodes: list) -> list[str]:
return [item for item in dict.fromkeys(items) if item]
def _normalize_publication_item(item: dict) -> dict:
def _normalize_publication_item(item: dict, current_author_id: str | None = None) -> dict:
publication_id = str(item.get("id") or "").strip()
title = _html_to_text(item.get("title"))
year = item.get("year")
year = _int_or_none(item.get("year"))
publication_type = str(item.get("type") or "").strip() or None
description = item.get("description") if isinstance(item.get("description"), dict) else {}
short_description = _localized_value(description.get("short")) or _localized_value(description.get("shortLeft"))
documents = item.get("documents") if isinstance(item.get("documents"), dict) else {}
language = item.get("language") if isinstance(item.get("language"), dict) else {}
annotation = _localized_text_map(item.get("annotation"))
authors = _normalize_publication_authors(item.get("authorsByType"), current_author_id)
citation_text = normalize_ws(str(description.get("main") or "")) or _build_publication_citation(title, authors, year)
text = normalize_ws(" ".join(part for part in [title, str(year or ""), short_description] if part))
return {
"id": publication_id or None,
"publication_id": publication_id or None,
"title": title or publication_id,
"year": year,
"type": publication_type,
"publication_type": publication_type,
"language": normalize_ws(language.get("name")) or None,
"status": _int_or_none(item.get("status")),
"url": f"https://publications.hse.ru/view/{publication_id}" if publication_id else None,
"doi_url": _document_href(documents, "DOI"),
"other_url": _document_href(documents, "OTHER_URL"),
"document_url": _document_href(documents, "DOCUMENT"),
"citation_text": citation_text or None,
"annotation": annotation,
"description": description or None,
"authors": authors,
"raw_data": item,
"text": text or title or publication_id,
}
@@ -598,12 +702,69 @@ def _html_to_text(value: object) -> str:
return normalize_ws(BeautifulSoup(str(value or ""), "html.parser").get_text(" ", strip=True))
def _localized_text_map(value: object) -> dict[str, str]:
if not isinstance(value, dict):
return {}
localized = {}
for key in ("ru", "en", "publ"):
text = _html_to_text(value.get(key))
if text:
localized[key] = text
return localized
def _localized_value(value: object) -> str:
if isinstance(value, dict):
return normalize_ws(value.get("ru") or value.get("publ") or value.get("en"))
return normalize_ws(str(value or ""))
def _normalize_publication_authors(value: object, current_author_id: str | None) -> list[dict]:
if not isinstance(value, dict):
return []
authors = []
for author in value.get("author") or []:
if not isinstance(author, dict):
continue
title = author.get("title") if isinstance(author.get("title"), dict) else {}
reverse_title = author.get("reverseTitle") if isinstance(author.get("reverseTitle"), dict) else {}
author_id = normalize_ws(author.get("id"))
href = normalize_ws(author.get("href"))
authors.append(
{
"id": author_id or None,
"href": urljoin("https://www.hse.ru", href) if href else None,
"title_ru": _html_to_text(title.get("ru")),
"title_en": _html_to_text(title.get("en")),
"reverse_title_ru": _html_to_text(reverse_title.get("ru")),
"reverse_title_en": _html_to_text(reverse_title.get("en")),
"alt_name": normalize_ws(author.get("altName")) or None,
"other_name": normalize_ws(author.get("otherName")) or None,
"is_current_employee": bool(current_author_id and author_id == current_author_id),
}
)
return authors
def _document_href(documents: dict, key: str) -> str | None:
document = documents.get(key)
if not isinstance(document, dict):
return None
return normalize_ws(document.get("href")) or None
def _build_publication_citation(title: str, authors: list[dict], year: int | None) -> str:
author_names = [author.get("title_ru") or author.get("title_en") or author.get("alt_name") for author in authors]
return normalize_ws(". ".join(part for part in [", ".join(filter(None, author_names)), title, str(year or "")] if part))
def _int_or_none(value: object) -> int | None:
try:
return int(value)
except (TypeError, ValueError):
return None
def _slugify(value: str) -> str:
cleaned = re.sub(r"[^\w\s-]", "", value.lower(), flags=re.UNICODE)
return re.sub(r"[-\s]+", "_", cleaned).strip("_") or "section"
@@ -629,3 +790,62 @@ def _dedupe_dicts(items: list[dict]) -> list[dict]:
seen.add(key)
unique.append(item)
return unique
def _fetch_text(
session: Session,
url: str,
headers: dict[str, str],
timeout: int,
*,
resource_cache=None,
profile_key: str | None = None,
resource_key: str,
resource_manifest: list[dict] | None,
method: str = "GET",
json_payload: object | None = None,
params: dict | None = None,
) -> str:
if resource_cache and profile_key:
cached = resource_cache.fetch_text(
session,
profile_key=profile_key,
resource_key=resource_key,
method=method,
url=url,
headers=headers,
timeout=timeout,
json_payload=json_payload,
params=params,
)
if resource_manifest is not None:
resource_manifest.append(
{
"resource_key": resource_key,
"method": method,
"url": url,
"body_hash": cached.body_hash,
"from_cache": cached.from_cache,
"status_code": cached.status_code,
}
)
return cached.text
if method.upper() == "POST":
response = session.post(url, json=json_payload, headers=headers, timeout=timeout, params=params)
else:
response = session.get(url, headers=headers, timeout=timeout, params=params)
response.raise_for_status()
text = response.text
if resource_manifest is not None:
resource_manifest.append(
{
"resource_key": resource_key,
"method": method,
"url": url,
"body_hash": hashlib.sha256(text.encode("utf-8")).hexdigest(),
"from_cache": False,
"status_code": response.status_code,
}
)
return text

View File

@@ -153,7 +153,7 @@ def stats_payload(db: Session) -> dict[str, Any]:
def run_payload(run: CrawlRun | None) -> dict[str, Any] | None:
if not run:
return None
processed = run.parsed_count + run.error_count
processed = run.parsed_count + run.skipped_count + run.error_count
percent = round((processed / run.found_count) * 100, 1) if run.found_count else 0
return {
"id": run.id,
@@ -166,6 +166,7 @@ def run_payload(run: CrawlRun | None) -> dict[str, Any] | None:
"finished_display": format_admin_datetime(run.finished_at),
"found_count": run.found_count,
"parsed_count": run.parsed_count,
"skipped_count": run.skipped_count,
"new_count": run.new_count,
"error_count": run.error_count,
"dismissed_count": run.dismissed_count,

View File

@@ -1,18 +1,30 @@
import gzip
import hashlib
import json
import re
import time
from datetime import datetime, timezone
import requests
from sqlalchemy import select
from sqlalchemy import inspect, select
from sqlalchemy.orm import Session
from app.config import Settings
from app.models import CrawlError, CrawlRun, CrawlRunEmployeeChange, Employee, EmployeeSnapshot, ParserSource, ProfileTab
from app.models import (
CrawlError,
CrawlRun,
CrawlRunEmployeeChange,
Employee,
EmployeePublication,
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.dataset_versions import get_or_create_current_version
from app.services.resource_cache import ResourceCache
HEADERS = {
"User-Agent": "Mozilla/5.0 (compatible; MIEMEmployeesBot/0.1.0; +https://miem.hse.ru/)"
@@ -28,8 +40,10 @@ def run_crawl(db: Session, settings: Settings) -> CrawlRun:
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]
@@ -47,12 +61,17 @@ def run_crawl(db: Session, settings: Settings) -> CrawlRun:
HEADERS,
settings.request_timeout,
settings.parser_use_playwright,
resource_cache=resource_cache,
)
if not parsed:
continue
_upsert_employee(db, run, parsed)
_, changed = _upsert_employee(db, run, parsed)
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
@@ -70,6 +89,7 @@ def run_crawl(db: Session, settings: Settings) -> CrawlRun:
run.dismissed_count = _mark_dismissed(db, run, found_keys, session, settings.request_timeout)
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)
@@ -88,21 +108,27 @@ def refresh_employee(db: Session, employee: Employee, settings: Settings) -> Cra
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("Распарсенный профиль не совпадает с обновляемым сотрудником.")
_upsert_employee(db, run, parsed)
_, changed = _upsert_employee(db, run, parsed)
if changed:
run.parsed_count = 1
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
@@ -137,8 +163,9 @@ 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) -> Employee:
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))
@@ -157,11 +184,14 @@ def _upsert_employee(db: Session, run: CrawlRun, parsed: dict) -> Employee:
else:
is_new = False
parser_version = parsed.get("parser_version")
changed = is_new or employee.current_checksum != checksum or employee.parser_version != parser_version
employee.full_name = parsed.get("full_name")
employee.status = "active"
employee.last_seen_at = now
employee.dismissed_at = None
employee.parser_version = parsed.get("parser_version")
employee.parser_version = parser_version
if changed:
employee.current_data = parsed
employee.current_checksum = checksum
db.flush()
@@ -176,6 +206,7 @@ def _upsert_employee(db: Session, run: CrawlRun, parsed: dict) -> Employee:
message="Сотрудник впервые найден в источнике.",
)
if changed:
db.query(ProfileTab).filter(ProfileTab.employee_id == employee.id).delete()
for tab in parsed.get("tabs") or []:
db.add(
@@ -194,10 +225,128 @@ def _upsert_employee(db: Session, run: CrawlRun, parsed: dict) -> Employee:
parsed_data=parsed,
html_snapshot=gzip.compress(html.encode("utf-8")) if html else None,
checksum=checksum,
parser_version=parsed.get("parser_version"),
parser_version=parser_version,
)
)
return employee
db.flush()
_try_sync_employee_publications(db, run, employee, parsed)
return employee, changed
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 _mark_dismissed(db: Session, run: CrawlRun, found_keys: set[str], session: requests.Session, timeout: int) -> int:
@@ -265,5 +414,23 @@ def _record_employee_change(
def _checksum(data: dict) -> str:
payload = json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
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,
)

View File

@@ -0,0 +1,227 @@
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

View File

@@ -0,0 +1,147 @@
from __future__ import annotations
import gzip
import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
import requests
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models import ParseResourceCache
from app.version import BACKEND_VERSION
@dataclass(frozen=True)
class CachedResource:
text: str
body_hash: str
from_cache: bool
status_code: int
class ResourceCache:
def __init__(self, db: Session):
self.db = db
def fetch_text(
self,
session: requests.Session,
*,
profile_key: str,
resource_key: str,
method: str,
url: str,
headers: dict[str, str],
timeout: int,
json_payload: Any | None = None,
params: dict[str, Any] | None = None,
) -> CachedResource:
method = method.upper()
fingerprint = _request_fingerprint(method=method, url=url, json_payload=json_payload, params=params)
cached = self.db.scalar(
select(ParseResourceCache).where(
ParseResourceCache.profile_key == profile_key,
ParseResourceCache.resource_key == resource_key,
ParseResourceCache.request_fingerprint == fingerprint,
)
)
request_headers = dict(headers)
if cached:
if cached.etag:
request_headers["If-None-Match"] = cached.etag
if cached.last_modified:
request_headers["If-Modified-Since"] = cached.last_modified
response = _send(
session,
method=method,
url=url,
headers=request_headers,
timeout=timeout,
json_payload=json_payload,
params=params,
)
if response.status_code == 304 and cached:
cached.fetched_at = datetime.now(timezone.utc)
self.db.flush()
return CachedResource(
text=gzip.decompress(cached.body_snapshot).decode("utf-8"),
body_hash=cached.body_hash,
from_cache=True,
status_code=response.status_code,
)
response.raise_for_status()
text = response.text
body_hash = _body_hash(text)
etag = response.headers.get("ETag") if hasattr(response, "headers") else None
last_modified = response.headers.get("Last-Modified") if hasattr(response, "headers") else None
if cached:
cached.method = method
cached.url = url
cached.etag = etag
cached.last_modified = last_modified
cached.body_hash = body_hash
cached.body_snapshot = gzip.compress(text.encode("utf-8"))
cached.parser_version = BACKEND_VERSION
cached.fetched_at = datetime.now(timezone.utc)
else:
self.db.add(
ParseResourceCache(
profile_key=profile_key,
resource_key=resource_key,
method=method,
url=url,
request_fingerprint=fingerprint,
etag=etag,
last_modified=last_modified,
body_hash=body_hash,
body_snapshot=gzip.compress(text.encode("utf-8")),
parser_version=BACKEND_VERSION,
fetched_at=datetime.now(timezone.utc),
)
)
self.db.flush()
return CachedResource(text=text, body_hash=body_hash, from_cache=False, status_code=response.status_code)
def _send(
session: requests.Session,
*,
method: str,
url: str,
headers: dict[str, str],
timeout: int,
json_payload: Any | None,
params: dict[str, Any] | None,
) -> requests.Response:
if method == "POST":
return session.post(url, json=json_payload, headers=headers, timeout=timeout, params=params)
return session.get(url, headers=headers, timeout=timeout, params=params)
def _request_fingerprint(
*,
method: str,
url: str,
json_payload: Any | None,
params: dict[str, Any] | None,
) -> str:
payload = {
"method": method,
"url": url,
"json": json_payload,
"params": params,
}
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
def _body_hash(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()

View File

@@ -89,12 +89,14 @@
const status = document.querySelector("[data-progress-status]");
const processed = document.querySelector("[data-progress-processed]");
const found = document.querySelector("[data-progress-found]");
const skipped = document.querySelector("[data-progress-skipped]");
const errors = document.querySelector("[data-progress-errors]");
const fill = document.querySelector("[data-progress-fill]");
const percent = document.querySelector("[data-progress-percent]");
if (status) status.textContent = run.status_display || run.status;
if (processed) processed.textContent = run.processed_count;
if (found) found.textContent = run.found_count;
if (skipped) skipped.textContent = run.skipped_count;
if (errors) errors.textContent = run.error_count;
if (fill) fill.style.width = `${run.progress_percent}%`;
if (percent) percent.textContent = run.progress_percent;

View File

@@ -37,6 +37,7 @@
<div class="progress-panel__meta">
<span data-progress-status>{{ run.status_display if run else "Ожидание" }}</span>
<span>обработано: <span data-progress-processed>{{ run.processed_count if run else 0 }}</span> / <span data-progress-found>{{ run.found_count if run else 0 }}</span></span>
<span>без изменений: <span data-progress-skipped>{{ run.skipped_count if run else 0 }}</span></span>
<span>ошибок: <span data-progress-errors>{{ run.error_count if run else 0 }}</span></span>
</div>
<div class="progress-bar" aria-label="Parsing progress">
@@ -48,10 +49,10 @@
<section class="panel">
<h2 class="panel__title">Последние запуски</h2>
<table class="table">
<thead><tr><th class="table__head">ID</th><th class="table__head">Статус</th><th class="table__head">Обработано</th><th class="table__head">Ошибки</th><th class="table__head">Старт</th></tr></thead>
<thead><tr><th class="table__head">ID</th><th class="table__head">Статус</th><th class="table__head">Обработано</th><th class="table__head">Без изменений</th><th class="table__head">Ошибки</th><th class="table__head">Старт</th></tr></thead>
<tbody>
{% for run in runs %}
<tr class="table__row" onclick="window.location.href='/admin/runs/{{ run.id }}'" onkeydown="if (event.key === 'Enter' || event.key === ' ') { event.preventDefault(); window.location.href='/admin/runs/{{ run.id }}'; }" role="link" tabindex="0"><td class="table__cell">{{ run.id }}</td><td class="table__cell">{{ run.status_display }}</td><td class="table__cell">{{ run.parsed_count }}</td><td class="table__cell">{{ run.error_count }}</td><td class="table__cell">{{ run.started_display }}</td></tr>
<tr class="table__row" onclick="window.location.href='/admin/runs/{{ run.id }}'" onkeydown="if (event.key === 'Enter' || event.key === ' ') { event.preventDefault(); window.location.href='/admin/runs/{{ run.id }}'; }" role="link" tabindex="0"><td class="table__cell">{{ run.id }}</td><td class="table__cell">{{ run.status_display }}</td><td class="table__cell">{{ run.parsed_count }}</td><td class="table__cell">{{ run.skipped_count }}</td><td class="table__cell">{{ run.error_count }}</td><td class="table__cell">{{ run.started_display }}</td></tr>
{% endfor %}
</tbody>
</table>

View File

@@ -12,6 +12,7 @@
<div class="stats-strip">
<div class="stats-strip__item"><span class="stats-strip__label">Найдено</span><span class="stats-strip__value">{{ run.found_count }}</span></div>
<div class="stats-strip__item"><span class="stats-strip__label">Обработано</span><span class="stats-strip__value">{{ run.parsed_count }}</span></div>
<div class="stats-strip__item"><span class="stats-strip__label">Без изменений</span><span class="stats-strip__value">{{ run.skipped_count }}</span></div>
<div class="stats-strip__item"><span class="stats-strip__label">Новые</span><span class="stats-strip__value">{{ run.new_count }}</span></div>
<div class="stats-strip__item"><span class="stats-strip__label">Потеряшки</span><span class="stats-strip__value">{{ run.changes.missing_from_source | length }}</span></div>
<div class="stats-strip__item"><span class="stats-strip__label">Уволены</span><span class="stats-strip__value">{{ run.dismissed_count }}</span></div>

View File

@@ -8,12 +8,13 @@
</div>
{% set run = runs[0] if runs else none %}
{% if run %}
{% set processed = run.parsed_count + run.error_count %}
{% set processed = run.parsed_count + run.skipped_count + run.error_count %}
{% set percent = ((processed / run.found_count) * 100) | round(1) if run.found_count else 0 %}
<div class="progress-panel" data-progress-panel>
<div class="progress-panel__meta">
<span data-progress-status>{{ run.status_display }}</span>
<span>обработано: <span data-progress-processed>{{ processed }}</span> / <span data-progress-found>{{ run.found_count }}</span></span>
<span>без изменений: <span data-progress-skipped>{{ run.skipped_count }}</span></span>
<span>ошибок: <span data-progress-errors>{{ run.error_count }}</span></span>
</div>
<div class="progress-bar" aria-label="Parsing progress">
@@ -26,6 +27,7 @@
<div class="progress-panel__meta">
<span data-progress-status>Ожидание</span>
<span>обработано: <span data-progress-processed>0</span> / <span data-progress-found>0</span></span>
<span>без изменений: <span data-progress-skipped>0</span></span>
<span>ошибок: <span data-progress-errors>0</span></span>
</div>
<div class="progress-bar" aria-label="Parsing progress">
@@ -35,10 +37,10 @@
</div>
{% endif %}
<table class="table">
<thead><tr><th class="table__head">ID</th><th class="table__head">Статус</th><th class="table__head">Найдено</th><th class="table__head">Обработано</th><th class="table__head">Новые</th><th class="table__head">Ошибки</th><th class="table__head">Уволены</th><th class="table__head">Старт</th></tr></thead>
<thead><tr><th class="table__head">ID</th><th class="table__head">Статус</th><th class="table__head">Найдено</th><th class="table__head">Обработано</th><th class="table__head">Без изменений</th><th class="table__head">Новые</th><th class="table__head">Ошибки</th><th class="table__head">Уволены</th><th class="table__head">Старт</th></tr></thead>
<tbody>
{% for run in runs %}
<tr class="table__row" onclick="window.location.href='/admin/runs/{{ run.id }}'" onkeydown="if (event.key === 'Enter' || event.key === ' ') { event.preventDefault(); window.location.href='/admin/runs/{{ run.id }}'; }" role="link" tabindex="0"><td class="table__cell">{{ run.id }}</td><td class="table__cell">{{ run.status_display }}</td><td class="table__cell">{{ run.found_count }}</td><td class="table__cell">{{ run.parsed_count }}</td><td class="table__cell">{{ run.new_count }}</td><td class="table__cell">{{ run.error_count }}</td><td class="table__cell">{{ run.dismissed_count }}</td><td class="table__cell">{{ run.started_display }}</td></tr>
<tr class="table__row" onclick="window.location.href='/admin/runs/{{ run.id }}'" onkeydown="if (event.key === 'Enter' || event.key === ' ') { event.preventDefault(); window.location.href='/admin/runs/{{ run.id }}'; }" role="link" tabindex="0"><td class="table__cell">{{ run.id }}</td><td class="table__cell">{{ run.status_display }}</td><td class="table__cell">{{ run.found_count }}</td><td class="table__cell">{{ run.parsed_count }}</td><td class="table__cell">{{ run.skipped_count }}</td><td class="table__cell">{{ run.new_count }}</td><td class="table__cell">{{ run.error_count }}</td><td class="table__cell">{{ run.dismissed_count }}</td><td class="table__cell">{{ run.started_display }}</td></tr>
{% endfor %}
</tbody>
</table>

View File

@@ -1,3 +1,3 @@
APP_VERSION = "0.4.7"
FRONTEND_VERSION = "0.4.7"
BACKEND_VERSION = "0.4.7"
APP_VERSION = "0.6.2"
FRONTEND_VERSION = "0.6.2"
BACKEND_VERSION = "0.6.2"

View File

@@ -17,7 +17,14 @@ def crawl_once() -> None:
settings = get_settings()
with SessionLocal() as db:
run = run_crawl(db, settings)
logger.info("crawl finished: id=%s status=%s parsed=%s errors=%s", run.id, run.status, run.parsed_count, run.error_count)
logger.info(
"crawl finished: id=%s status=%s parsed=%s skipped=%s errors=%s",
run.id,
run.status,
run.parsed_count,
run.skipped_count,
run.error_count,
)
def main() -> None:

View File

@@ -13,6 +13,7 @@ CREATE TABLE IF NOT EXISTS crawl_runs (
finished_at TIMESTAMPTZ,
found_count INTEGER NOT NULL DEFAULT 0,
parsed_count INTEGER NOT NULL DEFAULT 0,
skipped_count INTEGER NOT NULL DEFAULT 0,
new_count INTEGER NOT NULL DEFAULT 0,
error_count INTEGER NOT NULL DEFAULT 0,
dismissed_count INTEGER NOT NULL DEFAULT 0,
@@ -73,3 +74,22 @@ CREATE TABLE IF NOT EXISTS profile_tabs (
);
CREATE INDEX IF NOT EXISTS ix_profile_tabs_employee_id ON profile_tabs (employee_id);
CREATE TABLE IF NOT EXISTS parse_resource_cache (
id SERIAL PRIMARY KEY,
profile_key VARCHAR(255) NOT NULL,
resource_key VARCHAR(255) NOT NULL,
method VARCHAR(16) NOT NULL,
url TEXT NOT NULL,
request_fingerprint VARCHAR(64) NOT NULL,
etag TEXT,
last_modified TEXT,
body_hash VARCHAR(64) NOT NULL,
body_snapshot BYTEA NOT NULL,
parser_version VARCHAR(32),
fetched_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT uq_parse_resource_cache_resource UNIQUE (profile_key, resource_key, request_fingerprint)
);
CREATE INDEX IF NOT EXISTS ix_parse_resource_cache_profile_key
ON parse_resource_cache (profile_key);

View File

@@ -0,0 +1,29 @@
CREATE TABLE IF NOT EXISTS dataset_versions (
id SERIAL PRIMARY KEY,
hash VARCHAR(64) NOT NULL UNIQUE,
previous_hash VARCHAR(64),
crawl_run_id INTEGER REFERENCES crawl_runs(id),
employee_count INTEGER NOT NULL DEFAULT 0,
active_count INTEGER NOT NULL DEFAULT 0,
dismissed_count INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS ix_dataset_versions_created_at
ON dataset_versions (created_at);
CREATE TABLE IF NOT EXISTS dataset_version_items (
id SERIAL PRIMARY KEY,
dataset_version_id INTEGER NOT NULL REFERENCES dataset_versions(id),
profile_key VARCHAR(255) NOT NULL,
employee_id INTEGER REFERENCES employees(id),
status VARCHAR(32) NOT NULL,
checksum VARCHAR(64) NOT NULL,
CONSTRAINT uq_dataset_version_items_version_profile UNIQUE (dataset_version_id, profile_key)
);
CREATE INDEX IF NOT EXISTS ix_dataset_version_items_hash
ON dataset_version_items (dataset_version_id);
CREATE INDEX IF NOT EXISTS ix_dataset_version_items_profile_key
ON dataset_version_items (profile_key);

View File

@@ -0,0 +1,21 @@
ALTER TABLE crawl_runs
ADD COLUMN IF NOT EXISTS skipped_count INTEGER NOT NULL DEFAULT 0;
CREATE TABLE IF NOT EXISTS parse_resource_cache (
id SERIAL PRIMARY KEY,
profile_key VARCHAR(255) NOT NULL,
resource_key VARCHAR(255) NOT NULL,
method VARCHAR(16) NOT NULL,
url TEXT NOT NULL,
request_fingerprint VARCHAR(64) NOT NULL,
etag TEXT,
last_modified TEXT,
body_hash VARCHAR(64) NOT NULL,
body_snapshot BYTEA NOT NULL,
parser_version VARCHAR(32),
fetched_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT uq_parse_resource_cache_resource UNIQUE (profile_key, resource_key, request_fingerprint)
);
CREATE INDEX IF NOT EXISTS ix_parse_resource_cache_profile_key
ON parse_resource_cache (profile_key);

View File

@@ -0,0 +1,39 @@
CREATE TABLE IF NOT EXISTS employee_publications (
id SERIAL PRIMARY KEY,
employee_id INTEGER NOT NULL REFERENCES employees(id) ON DELETE CASCADE,
publication_id VARCHAR(64),
title TEXT NOT NULL,
year INTEGER,
publication_type VARCHAR(64),
language VARCHAR(16),
status INTEGER,
url TEXT,
doi_url TEXT,
other_url TEXT,
document_url TEXT,
citation_text TEXT,
annotation JSONB,
description JSONB,
authors JSONB,
raw_data JSONB,
source_hash VARCHAR(64) NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
CONSTRAINT uq_employee_publications_employee_publication UNIQUE (employee_id, publication_id),
CONSTRAINT uq_employee_publications_employee_source_hash UNIQUE (employee_id, source_hash)
);
CREATE INDEX IF NOT EXISTS ix_employee_publications_employee_id
ON employee_publications (employee_id);
CREATE INDEX IF NOT EXISTS ix_employee_publications_publication_id
ON employee_publications (publication_id);
CREATE INDEX IF NOT EXISTS ix_employee_publications_doi_url
ON employee_publications (doi_url);
CREATE INDEX IF NOT EXISTS ix_employee_publications_year
ON employee_publications (year);
CREATE INDEX IF NOT EXISTS ix_employee_publications_publication_type
ON employee_publications (publication_type);

View File

@@ -1,6 +1,6 @@
[project]
name = "miem-workers"
version = "0.4.7"
version = "0.6.2"
description = "MIEM employees parser, admin API, and MCP server"
requires-python = ">=3.11"
dependencies = [

View File

@@ -200,13 +200,14 @@ def test_run_payload_calculates_progress():
status="running",
found_count=10,
parsed_count=4,
skipped_count=2,
error_count=1,
)
payload = run_payload(run)
assert payload["processed_count"] == 5
assert payload["progress_percent"] == 50.0
assert payload["processed_count"] == 7
assert payload["progress_percent"] == 70.0
assert payload["status_display"] == "Выполняется"

View File

@@ -1,3 +1,4 @@
import json
from datetime import datetime, timezone
from types import SimpleNamespace
@@ -9,7 +10,7 @@ from sqlalchemy.pool import StaticPool
from app.config import Settings, get_settings
from app.db import Base, get_db
from app.main import app
from app.models import CrawlRun, CrawlRunEmployeeChange, Employee
from app.models import CrawlRun, CrawlRunEmployeeChange, Employee, EmployeePublication
from app.security import SESSION_COOKIE, sign_session
@@ -19,7 +20,7 @@ def test_health_returns_versions():
response = client.get("/api/health")
assert response.status_code == 200
assert response.json()["backend_version"] == "0.4.7"
assert response.json()["backend_version"] == "0.6.2"
def test_mcp_lists_tools_without_auth_and_ignores_auth_header():
@@ -50,7 +51,10 @@ def test_mcp_lists_tools_without_auth_and_ignores_auth_header():
assert without_auth.status_code == 200
assert with_auth.status_code == 200
assert without_auth.json()["result"]["tools"][0]["name"] == "search_employees"
tool_names = {tool["name"] for tool in without_auth.json()["result"]["tools"]}
assert "search_employees" in tool_names
assert "get_service_info" in tool_names
assert "sync_employees" in tool_names
assert any(tool["name"] == "get_crawl_run_details" for tool in without_auth.json()["result"]["tools"])
assert with_auth.json()["result"]["tools"] == without_auth.json()["result"]["tools"]
@@ -108,6 +112,230 @@ def test_mcp_search_employees_returns_matching_employee():
app.dependency_overrides.clear()
def test_mcp_service_info_returns_tools_and_dataset_hash():
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.add(
Employee(
profile_key="staff:alpha",
profile_type="staff",
profile_id="alpha",
canonical_url="https://www.hse.ru/staff/alpha",
full_name="Alpha Person",
status="active",
current_checksum="a" * 64,
current_data={"sections": []},
)
)
session.commit()
session.close()
def override_db():
db = Session()
try:
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_db
client = TestClient(app)
response = client.post(
"/mcp",
json={"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "get_service_info", "arguments": {}}},
)
assert response.status_code == 200
payload = json.loads(response.json()["result"]["content"][0]["text"])
assert payload["service_name"] == "miem-employees"
assert payload["backend_version"] == "0.6.2"
assert payload["dataset"]["hash"]
assert any(tool["name"] == "sync_employees" for tool in payload["tools"])
app.dependency_overrides.clear()
def test_mcp_list_employee_publications_prefers_stored_publications_with_fallback():
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
stored_employee = Employee(
profile_key="staff:stored",
profile_type="staff",
profile_id="stored",
canonical_url="https://www.hse.ru/staff/stored",
full_name="Stored Person",
status="active",
current_data={
"sections": [
{
"type": "publications",
"publications": [{"title": "Old JSON Publication", "url": "https://example.test/old"}],
}
]
},
)
fallback_employee = Employee(
profile_key="staff:fallback",
profile_type="staff",
profile_id="fallback",
canonical_url="https://www.hse.ru/staff/fallback",
full_name="Fallback Person",
status="active",
current_data={
"sections": [
{
"type": "publications",
"publications": [{"title": "Fallback Publication", "url": "https://example.test/fallback"}],
}
]
},
)
session.add_all([stored_employee, fallback_employee])
session.commit()
session.add(
EmployeePublication(
employee_id=stored_employee.id,
publication_id="pub-1",
title="Stored Publication",
year=2024,
publication_type="ARTICLE",
url="https://publications.hse.ru/view/pub-1",
doi_url="https://doi.org/10.1/test",
citation_text="Stored Citation",
annotation={"ru": "Аннотация", "en": "Abstract"},
description={"main": "Stored Citation"},
authors=[{"id": "1", "title_ru": "Автор", "is_current_employee": True}],
source_hash="a" * 64,
)
)
session.commit()
session.close()
def override_db():
db = Session()
try:
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_db
client = TestClient(app)
stored_response = client.post(
"/mcp",
json={
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {"name": "list_employee_publications", "arguments": {"profile_id_or_url": "stored"}},
},
)
fallback_response = client.post(
"/mcp",
json={
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {"name": "list_employee_publications", "arguments": {"profile_id_or_url": "fallback"}},
},
)
stored_payload = json.loads(stored_response.json()["result"]["content"][0]["text"])
fallback_payload = json.loads(fallback_response.json()["result"]["content"][0]["text"])
assert stored_payload["items"][0]["title"] == "Stored Publication"
assert stored_payload["items"][0]["doi_url"] == "https://doi.org/10.1/test"
assert stored_payload["items"][0]["annotation"] == {"ru": "Аннотация", "en": "Abstract"}
assert stored_payload["items"][0]["authors"] == [{"id": "1", "title_ru": "Автор", "is_current_employee": True}]
assert fallback_payload["items"][0]["title"] == "Fallback Publication"
app.dependency_overrides.clear()
def test_mcp_sync_employees_full_empty_and_unknown_hash_modes():
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
session.add(
Employee(
profile_key="staff:alpha",
profile_type="staff",
profile_id="alpha",
canonical_url="https://www.hse.ru/staff/alpha",
full_name="Alpha Person",
status="active",
current_checksum="a" * 64,
current_data={"sections": [{"type": "paragraphs"}]},
)
)
session.commit()
session.close()
def override_db():
db = Session()
try:
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_db
client = TestClient(app)
full_response = client.post(
"/mcp",
json={"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "sync_employees", "arguments": {}}},
)
full_payload = json.loads(full_response.json()["result"]["content"][0]["text"])
current_hash = full_payload["to_hash"]
empty_response = client.post(
"/mcp",
json={
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {"name": "sync_employees", "arguments": {"client_hash": current_hash}},
},
)
empty_payload = json.loads(empty_response.json()["result"]["content"][0]["text"])
unknown_response = client.post(
"/mcp",
json={
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {"name": "sync_employees", "arguments": {"client_hash": "missing"}},
},
)
unknown_payload = json.loads(unknown_response.json()["result"]["content"][0]["text"])
assert full_payload["mode"] == "full"
assert full_payload["items"][0]["data"] == {"sections": [{"type": "paragraphs"}]}
assert empty_payload["mode"] == "delta"
assert empty_payload["changes"] == {"added": [], "updated": [], "dismissed": [], "removed": []}
assert unknown_payload["mode"] == "full"
assert unknown_payload["reason"] == "unknown_client_hash"
app.dependency_overrides.clear()
def test_mcp_get_crawl_run_details_returns_changes():
engine = create_engine(
"sqlite:///:memory:",

View File

@@ -1,7 +1,9 @@
import gzip
from datetime import datetime, timezone
from app.models import CrawlRun, CrawlRunEmployeeChange, Employee
from app.services.crawler import _mark_dismissed, _upsert_employee
from app.models import CrawlError, CrawlRun, CrawlRunEmployeeChange, Employee, EmployeePublication, EmployeeSnapshot, ParseResourceCache
from app.services.crawler import _checksum, _mark_dismissed, _upsert_employee
from app.services.resource_cache import ResourceCache
class FakeResponse:
@@ -17,6 +19,34 @@ class FakeSession:
return FakeResponse(self.statuses[url])
class ConditionalResponse:
def __init__(self, status_code, text="", headers=None):
self.status_code = status_code
self._text = text
self.headers = headers or {}
self.text_read = False
@property
def text(self):
self.text_read = True
return self._text
def raise_for_status(self):
return None
class ConditionalSession:
def __init__(self):
self.requests = []
self.not_modified_response = ConditionalResponse(304)
def get(self, url, **kwargs):
self.requests.append((url, kwargs))
if kwargs["headers"].get("If-None-Match") == '"cached"':
return self.not_modified_response
return ConditionalResponse(200, "fresh", {"ETag": '"fresh"'})
def test_mark_dismissed_records_missing_source_when_profile_is_available(db_session):
run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
db_session.add(run)
@@ -111,3 +141,176 @@ def test_upsert_employee_increments_new_count_and_records_change_for_new_employe
change = db_session.query(CrawlRunEmployeeChange).one()
assert change.change_type == "new"
assert change.full_name == "New Person"
def test_resource_cache_uses_etag_and_reuses_cached_body_on_304(db_session):
db_session.add(
ParseResourceCache(
profile_key="staff:cached",
resource_key="main-html",
method="GET",
url="https://www.hse.ru/staff/cached",
request_fingerprint="020d59db7b358d9023d0f185bcbf5a9c085d3cf2bf91d92d48eee9147e8d0f01",
etag='"cached"',
body_hash="cached-hash",
body_snapshot=gzip.compress("cached body".encode("utf-8")),
parser_version="0.6.0",
)
)
db_session.commit()
session = ConditionalSession()
result = ResourceCache(db_session).fetch_text(
session,
profile_key="staff:cached",
resource_key="main-html",
method="GET",
url="https://www.hse.ru/staff/cached",
headers={"User-Agent": "test"},
timeout=10,
)
assert session.requests[0][1]["headers"]["If-None-Match"] == '"cached"'
assert result.text == "cached body"
assert result.from_cache is True
assert session.not_modified_response.text_read is False
def test_upsert_employee_skips_snapshot_when_checksum_is_unchanged(db_session):
first_run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
second_run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
db_session.add_all([first_run, second_run])
db_session.commit()
_, first_changed = _upsert_employee(db_session, first_run, _parsed_employee("same"))
_, second_changed = _upsert_employee(db_session, second_run, _parsed_employee("same"))
db_session.commit()
assert first_changed is True
assert second_changed is False
assert db_session.query(EmployeeSnapshot).count() == 1
def test_upsert_employee_saves_publications_and_reuses_existing_rows(db_session):
first_run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
second_run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
db_session.add_all([first_run, second_run])
db_session.commit()
parsed = _parsed_employee("published")
parsed["sections"] = [
{
"type": "publications",
"publications": [
{
"id": "888959076",
"publication_id": "888959076",
"title": "Detailed Publication",
"year": 2023,
"publication_type": "ARTICLE",
"language": "ru",
"status": 1,
"url": "https://publications.hse.ru/view/888959076",
"doi_url": "https://doi.org/10.1/test",
"citation_text": "Detailed citation",
"annotation": {"ru": "Аннотация"},
"description": {"main": "Detailed citation"},
"authors": [{"id": "1", "title_ru": "Автор"}],
"raw_data": {"id": "888959076", "title": "Detailed Publication"},
}
],
}
]
employee, _ = _upsert_employee(db_session, first_run, parsed)
db_session.commit()
_upsert_employee(db_session, second_run, _parsed_employee_with_publication("published"))
db_session.commit()
publications = db_session.query(EmployeePublication).filter_by(employee_id=employee.id).all()
assert len(publications) == 1
assert publications[0].doi_url == "https://doi.org/10.1/test"
assert publications[0].authors == [{"id": "1", "title_ru": "Автор"}]
def test_upsert_employee_records_publication_errors_without_failing_employee(monkeypatch, db_session):
run = CrawlRun(source_url="https://miem.hse.ru/persons", status="running")
db_session.add(run)
db_session.commit()
def broken_sync(*_args, **_kwargs):
raise RuntimeError("boom")
monkeypatch.setattr("app.services.crawler._sync_employee_publications", broken_sync)
employee, changed = _upsert_employee(db_session, run, _parsed_employee_with_publication("error-safe"))
db_session.commit()
assert changed is True
assert employee.full_name == "Same Person"
assert db_session.query(Employee).filter_by(profile_key="staff:error-safe").one()
error = db_session.query(CrawlError).one()
assert "публикации" in error.message.lower()
def test_checksum_changes_when_widget_data_changes():
base = _parsed_employee("widgets")
changed = _parsed_employee("widgets")
changed["sections"] = [
{
"type": "publications",
"publications": [{"id": "1", "title": "New publication"}],
}
]
assert _checksum(base) != _checksum(changed)
def test_checksum_ignores_date_dependent_experience_text():
first = _parsed_employee("experience")
second = _parsed_employee("experience")
first["sections"] = [{"raw_text": "Стаж работы в НИУ ВШЭ: 5 лет"}]
second["sections"] = [{"raw_text": "Стаж работы в НИУ ВШЭ: 6 лет"}]
assert _checksum(first) == _checksum(second)
def _parsed_employee(profile_id: str) -> dict:
return {
"source_url": f"https://www.hse.ru/staff/{profile_id}",
"profile_type": "staff",
"profile_id": profile_id,
"full_name": "Same Person",
"tabs": [],
"sections": [],
"parser_version": "0.6.0",
"_html": "<html></html>",
}
def _parsed_employee_with_publication(profile_id: str) -> dict:
parsed = _parsed_employee(profile_id)
parsed["sections"] = [
{
"type": "publications",
"publications": [
{
"id": "888959076",
"publication_id": "888959076",
"title": "Detailed Publication",
"year": 2023,
"publication_type": "ARTICLE",
"language": "ru",
"status": 1,
"url": "https://publications.hse.ru/view/888959076",
"doi_url": "https://doi.org/10.1/test",
"citation_text": "Detailed citation",
"annotation": {"ru": "Аннотация"},
"description": {"main": "Detailed citation"},
"authors": [{"id": "1", "title_ru": "Автор"}],
"raw_data": {"id": "888959076", "title": "Detailed Publication"},
}
],
}
]
return parsed

View File

@@ -0,0 +1,88 @@
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"

71
tests/test_db_schema.py Normal file
View File

@@ -0,0 +1,71 @@
from sqlalchemy import create_engine, inspect, text
from app.db import _ensure_runtime_schema
def test_runtime_schema_adds_skipped_count_to_existing_crawl_runs_table(monkeypatch):
engine = create_engine("sqlite:///:memory:")
with engine.begin() as connection:
connection.execute(
text(
"""
CREATE TABLE crawl_runs (
id INTEGER PRIMARY KEY,
source_url TEXT NOT NULL,
status VARCHAR(32) NOT NULL DEFAULT 'running',
found_count INTEGER NOT NULL DEFAULT 0,
parsed_count INTEGER NOT NULL DEFAULT 0
)
"""
)
)
monkeypatch.setattr("app.db.engine", engine)
_ensure_runtime_schema()
columns = {column["name"] for column in inspect(engine).get_columns("crawl_runs")}
assert "skipped_count" in columns
def test_runtime_schema_creates_employee_publications_table_when_employees_exist(monkeypatch):
engine = create_engine("sqlite:///:memory:")
with engine.begin() as connection:
connection.execute(
text(
"""
CREATE TABLE employees (
id INTEGER PRIMARY KEY,
profile_key VARCHAR(255) NOT NULL UNIQUE,
canonical_url TEXT NOT NULL,
status VARCHAR(32) NOT NULL DEFAULT 'active',
first_seen_at DATETIME NOT NULL,
last_seen_at DATETIME NOT NULL,
created_at DATETIME NOT NULL,
updated_at DATETIME NOT NULL
)
"""
)
)
connection.execute(
text(
"""
CREATE TABLE crawl_runs (
id INTEGER PRIMARY KEY,
source_url TEXT NOT NULL,
status VARCHAR(32) NOT NULL DEFAULT 'running',
found_count INTEGER NOT NULL DEFAULT 0,
parsed_count INTEGER NOT NULL DEFAULT 0,
skipped_count INTEGER NOT NULL DEFAULT 0
)
"""
)
)
monkeypatch.setattr("app.db.engine", engine)
_ensure_runtime_schema()
_ensure_runtime_schema()
inspector = inspect(engine)
assert "employee_publications" in inspector.get_table_names()
columns = {column["name"] for column in inspector.get_columns("employee_publications")}
assert {"employee_id", "publication_id", "doi_url", "authors", "raw_data", "source_hash"}.issubset(columns)

View File

@@ -34,7 +34,21 @@ class FakeSession:
"type": "ARTICLE",
"title": "Дублирование пакетов",
"year": 2023,
"language": {"name": "ru"},
"status": 1,
"authorsByType": {
"author": [
{
"id": "568398853",
"href": "/org/persons/568398853",
"title": {"ru": "Левицкий И. А.", "en": ""},
"reverseTitle": {"ru": "И. А. Левицкий", "en": ""},
}
]
},
"description": {"short": {"ru": "Информационные процессы. 2023."}},
"annotation": {"ru": "<p>Русская аннотация</p>"},
"documents": {"DOI": {"href": "https://doi.org/10.1/test"}},
}
],
},
@@ -153,6 +167,9 @@ def test_enrich_sections_from_hse_widgets_loads_publications_and_vkr():
assert publications["publications_count"] == 1
assert publications["publications"][0]["url"] == "https://publications.hse.ru/view/888959076"
assert publications["publications"][0]["doi_url"] == "https://doi.org/10.1/test"
assert publications["publications"][0]["annotation"] == {"ru": "Русская аннотация"}
assert publications["publications"][0]["authors"][0]["is_current_employee"] is True
assert theses["theses_count"] == 1
assert theses["theses"][0]["student"] == "Лесняк Владислав Евгеньевич"
assert theses["theses"][0]["project_url"] == "https://www.hse.ru/edu/vkr/1045750164"