fix: respect publication types and isolate catalog failures

CHANGELOG.md

@@ -1,5 +1,10 @@
 # Changelog
 
+## 0.16.1
+
+- Ручные основания больше не возвращают книги и другие недопущенные виды в предварительные счётчики совета. Дополнительные виды учитываются только при подтверждённой разрешающей политике и согласованном виде материала.
+- Повреждение одного квартильного каталога не блокирует метрики исправного второго. Общее предупреждение о снимке отделено от проблем конкретной публикации; страница и XLSX сохраняют доступные проверенные сведения.
+
 ## 0.16.0
 
 - Проверка публикационных критериев совета расширена на кандидатов наук и PhD. Выбирается высшая степень из истории; сотрудники без публикаций остаются видимыми. Добавлены дата ходатайства, отрасль оценки, пороги пунктов 5–6 и отдельный минимум монографий.

README.md

@@ -155,7 +155,7 @@ node --check app/static/admin.js
 
 Тесты с данными используют временную SQLite; отдельные API smoke-тесты запускают приложение с его текущей конфигурацией. Браузерная проверка: `pip install playwright`, `python -m playwright install chromium`, затем `python tests/browser_admin.py`. Она проверяет реальный рендеринг страниц, клавиатуру, диалог, фильтры и восстановление прогресса после ошибки. Скриншоты сохраняются во временную папку; путь выводится в конце.
 
-Версия сервиса: `0.16.0`. Админка всегда показывает версии backend и frontend в footer.
+Версия сервиса: `0.16.1`. Админка всегда показывает версии backend и frontend в footer.
 
 ## Разовый импорт публикаций
 

app/services/council.py

@@ -19,7 +19,10 @@ from app.models import Employee
 from app.services.academic_degrees import academic_degrees
 from app.services.council_source import _doi, _journal_names, _name, _quartiles, load_reference
 from app.services.council_catalog import catalog, publication_catalog
-from app.services.council_assessment import BRANCHES, assess_candidate, degree_level, evidence, iso_date, publication_proofs
+from app.services.council_assessment import (
+    BRANCHES, assess_candidate, degree_level, evidence, iso_date, publication_proofs,
+    publication_type_allowed, sourced,
+)
 
 
 EXPORT = Path(__file__).resolve().parents[1] / "data" / "council_articles.json"
@@ -259,6 +262,13 @@ def council_candidates(db: Session, today: date | None = None, q: str = "", offs
         articles = _unique_articles([*csv_articles, *(_database_article(p) for p in employee.publications)], start, end)
         for article in articles:
             proofs = publication_proofs(employee_evidence, article)
+            kinds = {proof["kind"] for proof in proofs if proof.get("kind") and sourced(proof.get("kind_evidence"))}
+            kind = next(iter(kinds)) if len(kinds) == 1 else None
+            preliminary_type_allowed = len(kinds) <= 1 and publication_type_allowed(
+                article["types"], kind, verified.get("policy", {}), preliminary=True)
+            if not preliminary_type_allowed:
+                for flag in ("q1", "q2", "k1", "k2", "rsci"):
+                    article[flag] = False
             identifiers = [*article["issns"], *(value for proof in proofs for value in proof.get("issns", []))]
             for name in article["journals"]:
                 canonical = _k1_journals().get(_journal_key(name))
@@ -266,9 +276,11 @@ def council_candidates(db: Session, today: date | None = None, q: str = "", offs
             article["catalog_records"], article["catalog_issues"] = publication_catalog(identifiers, article["journals"], specialty)
             for record in article["catalog_records"]:
                 for metric in record["metrics"]:
-                    if article["year"] == metric["year"] and start.year <= article["year"] <= end.year and metric["quartile"] in {"Q1", "Q2"} and not article["catalog_issues"] and (not article["types"] or article["types"] == ["Статья"]):
+                    if article["year"] == metric["year"] and start.year <= article["year"] <= end.year and metric["quartile"] in {"Q1", "Q2"} and not article["catalog_issues"] and preliminary_type_allowed:
                         article[metric["quartile"].lower()] = True
             for proof in proofs:
+                if not preliminary_type_allowed:
+                    continue
                 for basis in proof.get("bases", []):
                     if not isinstance(basis, dict) or basis.get("applicable_year") != article["year"] or article["year"] is None or not start.year <= article["year"] <= end.year:
                         continue

app/services/council_assessment.py

@@ -21,6 +21,8 @@ for names, minimum, monographs in [
     for name in names.split("|"):
         BRANCHES[name] = (minimum, monographs)
 SYSTEMS = {"Web of Science", "Scopus", "PubMed", "MathSciNet", "zbMATH", "Chemical Abstracts", "Springer", "GeoRef"}
+TYPE_LABELS = {"article": "Статья", "monograph": "Книга", "book": "Книга", "chapter": "Глава книги",
+               "report": "Доклад", "patent": "Патент", "preprint": "Препринт", "conference_paper": "Материал конференции"}
 
 
 def iso_date(value):
@@ -39,6 +41,17 @@ def sourced(record):
     return True
 
 
+def publication_type_allowed(types, kind, policy, *, preliminary=False):
+    if kind is None:
+        return preliminary and (not types or types == ["Статья"])
+    if types and types != [TYPE_LABELS.get(kind)]:
+        return False
+    if kind == "article":
+        return True
+    admitted = policy.get("publication_types", {})
+    return sourced(admitted) and kind in admitted.get("values", [])
+
+
 def degree_level(degree):
     if re.fullmatch(r"доктор\b.+\bнаук", degree, re.I):
         return 2
@@ -144,17 +157,12 @@ def assess_article(article, proofs, start, end, level, branch, specialty, policy
     if len(types) != 1:
         reasons.append("Вид материала не подтверждён или расходится")
     kind = next(iter(types)) if len(types) == 1 else None
-    type_labels = {"article": "Статья", "monograph": "Книга", "book": "Книга", "chapter": "Глава книги",
-                   "report": "Доклад", "patent": "Патент", "preprint": "Препринт", "conference_paper": "Материал конференции"}
-    type_conflict = len(types) > 1 or bool(article["types"] and kind and article["types"] != [type_labels.get(kind)])
+    type_conflict = len(types) > 1 or bool(article["types"] and kind and article["types"] != [TYPE_LABELS.get(kind)])
     if type_conflict:
         reasons.append("Виды материала в источниках расходятся")
-    eligible_kind = kind == "article"
-    if kind and kind != "article":
-        admissible = policy.get("publication_types", {})
-        eligible_kind = sourced(admissible) and kind in admissible.get("values", [])
-        if not eligible_kind:
-            reasons.append("Зачёт этого вида материала в общий минимум не подтверждён")
+    eligible_kind = publication_type_allowed(article["types"], kind, policy)
+    if kind and kind != "article" and not eligible_kind:
+        reasons.append("Зачёт этого вида материала в общий минимум не подтверждён")
     bases = []
     basis_year = exact.year if exact else article["year"]
     for proof in proofs:

app/services/council_catalog.py

@@ -142,7 +142,9 @@ def catalog():
 def publication_catalog(identifiers, names, specialty=""):
     data = catalog()
     matches = set().union(*(data["by_issn"].get(issn(value), set()) for value in identifiers))
-    issues = list(data["issues"])
+    # Snapshot failures are reported once in the assessment context. Only
+    # problems with this journal may prevent use of its validated metrics.
+    issues = []
     if len(matches) > 1:
         return [], [*issues, "ISSN указывают на разные журналы — требуется проверка"]
     records = data["journals"].get(next(iter(matches)), []) if matches else []

app/version.py

@@ -1,3 +1,3 @@
-APP_VERSION = "0.16.0"
-FRONTEND_VERSION = "0.16.0"
-BACKEND_VERSION = "0.16.0"
+APP_VERSION = "0.16.1"
+FRONTEND_VERSION = "0.16.1"
+BACKEND_VERSION = "0.16.1"

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "miem-workers"
-version = "0.16.0"
+version = "0.16.1"
 description = "MIEM employees parser, admin API, and web admin"
 requires-python = ">=3.11"
 dependencies = [

tests/test_api.py

@@ -18,7 +18,7 @@ def test_health_returns_versions():
     response = TestClient(app).get("/api/health")
 
     assert response.status_code == 200
-    assert response.json()["backend_version"] == "0.16.0"
+    assert response.json()["backend_version"] == "0.16.1"
 
 
 def test_api_employees_and_stats_require_admin_session():

tests/test_council.py

@@ -218,7 +218,7 @@ def test_public_page_data_minimization_and_safe_links(db_session, export):
         assert "&lt;script&gt;" in response.text and "<script>" not in response.text
         assert "javascript:" not in response.text and "private@example.org" not in response.text
         assert "/admin/employees/" not in response.text and "/admin/logout" not in response.text
-        assert "Backend 0.16.0 · Frontend 0.16.0" in response.text
+        assert "Backend 0.16.1 · Frontend 0.16.1" in response.text
         assert "Вход в админку" in response.text
         assert client.get("/admin/directory", follow_redirects=False).status_code in (302, 303, 401, 403)
         assert "Сотрудники не найдены" in client.get("/admin/council?q=missing").text

tests/test_council_assessment.py

@@ -292,3 +292,75 @@ def test_invalid_evidence_schema_is_rejected(monkeypatch, tmp_path):
             assessment.evidence()
     finally:
         assessment.evidence.cache_clear()
+
+
+@pytest.mark.parametrize("basis,flag", [("К1", "k1"), ("К2", "k2"), ("RSCI", "rsci"), ("Q1", "q1"), ("Q2", "q2")])
+@pytest.mark.parametrize("kind", ["BOOK", None])
+@pytest.mark.parametrize("admitted", [False, True])
+def test_manual_monograph_bases_respect_preliminary_type_policy(db_session, proofs, basis, flag, kind, admitted):
+    employee = person(db_session, ["Доктор технических наук"])
+    record = work(db_session, employee, kind=kind, basis=basis)
+    record["kind"] = "monograph"
+    policy = {"publication_types": {**SOURCE, "values": ["monograph"]}} if admitted else {}
+    proofs(publications=[record], policy=policy)
+    full_result = council.council_candidates(db_session, petition_date=date(2026, 10, 8), branch="технические")
+    candidate = full_result["candidates"][0]
+    assert candidate[f"{flag}_count"] == candidate["total_count"] == int(admitted)
+    # The policy permits a type; it does not itself confirm a quartile.
+    assert candidate["confirmed_count"] == (int(admitted) if basis not in {"Q1", "Q2"} else 0)
+    book = load_workbook(BytesIO(council_xlsx(full_result)))
+    headers = [cell.value for cell in book["Сотрудники"][1]]
+    exported = dict(zip(headers, [cell.value for cell in book["Сотрудники"][2]]))
+    assert exported["ВСЕГО"] == int(admitted)
+    assert exported[{"К1": "К1", "К2": "К2 предварительно", "RSCI": "RSCI предварительно", "Q1": "Q1", "Q2": "Q2"}[basis]] == int(admitted)
+
+
+def test_admitted_type_needs_sourced_policy_and_consistent_material(db_session, proofs):
+    employee = person(db_session, ["Доктор технических наук"])
+    record = work(db_session, employee, kind="BOOK")
+    record["kind"] = "monograph"
+    proofs(publications=[record], policy={"publication_types": {"values": ["monograph"]}})
+    assert result(db_session)["total_count"] == 0
+    record["kind"] = "article"
+    proofs(publications=[record])
+    assert result(db_session)["total_count"] == 0
+
+
+@pytest.mark.parametrize("broken_index", [0, 1])
+def test_valid_snapshot_keeps_preliminary_metrics_when_other_is_broken(db_session, proofs, monkeypatch, tmp_path, broken_index):
+    from app.services import council_catalog as catalogs
+    paths = list(catalogs.CATALOGS)
+    valid_path = paths[1 - broken_index]
+    valid = json.loads(valid_path.read_text(encoding="utf8"))
+    journal = valid["journals"][0]
+    broken = tmp_path / "broken" / "catalog.json"
+    broken.parent.mkdir()
+    broken.write_text("{}", encoding="utf8")
+    paths[broken_index] = broken
+    monkeypatch.setattr(catalogs, "CATALOGS", tuple(paths))
+    catalogs.catalog.cache_clear()
+    try:
+        employee = person(db_session, ["Доктор технических наук"])
+        db_session.add(EmployeePublication(employee_id=employee.id, title="Snapshot work", year=2025,
+            publication_type="ARTICLE", source_hash="snapshot", raw_data={"journal": journal["name"], "issn": journal["issn"]}))
+        db_session.commit()
+        full_result = council.council_candidates(db_session, petition_date=date(2026, 10, 8), branch="технические")
+        candidate = full_result["candidates"][0]
+        article = candidate["articles"][0]
+        flag = "q2" if broken_index == 0 else "q1"
+        assert candidate[f"{flag}_count"] == candidate["total_count"] == 1
+        assert candidate["confirmed_count"] == 0
+        assert full_result["catalog_issues"] and not article["catalog_issues"]
+        assert {r["catalog"] for r in article["catalog_records"]} == {valid_path.parent.name}
+        book = load_workbook(BytesIO(council_xlsx(full_result)))
+        assert book["Сотрудники"].cell(2, 5 if flag == "q2" else 4).value == 1
+        assert "целостность каталога" in " ".join(str(cell.value) for row in book["Параметры"] for cell in row)
+        app.dependency_overrides[get_db] = lambda: db_session
+        try:
+            response = TestClient(app).get("/admin/council?petition_date=2026-10-08&branch=технические")
+            assert response.status_code == 200 and "целостность каталога" in response.text
+            assert journal["name"] in response.text
+        finally:
+            app.dependency_overrides.clear()
+    finally:
+        catalogs.catalog.cache_clear()