fix: preserve categorized quartile evidence for council assessment

app/services/council.py

@@ -17,7 +17,7 @@ from app.models import Employee, CouncilCorrections
 from app.services.academic_degrees import academic_degrees
 from app.services.council_source import _doi, _journal_names, _name, load_reference, _description_journals
 from app.services.council_catalog import catalog, publication_catalog
-from app.services.publication_quartiles import labels as quartile_labels, identity_issues, hse_entries, merge as merge_quartiles
+from app.services.publication_quartiles import qualifying_entries, identity_issues, hse_entries, merge as merge_quartiles
 from app.services.council_assessment import (
     BRANCHES, assess_candidate, degree_level, evidence, publication_proofs,
     publication_type_allowed, sourced,
@@ -120,10 +120,12 @@ def _database_article(publication) -> dict:
     if isinstance(imported, dict) and not imported_issues and (not imported_doi or imported_doi == doi):
         journals.extend(_journal_names(imported))
     issues = list(dict.fromkeys([*issues, *imported_issues]))
+    quartile_entries = [*qualifying_entries(publication.quartiles, publication.year, doi, current_journals, identifiers),
+                        *qualifying_entries(imported_quartiles, publication.year, doi, current_journals, identifiers)]
     return {"title": publication.title, "year": publication.year, "doi": _doi(publication.doi_url or ""),
             "types": [TYPES.get(kind.upper(), f"Другой тип: {kind}")] if kind else [],
-            "quartiles": list(dict.fromkeys([*quartile_labels(publication.quartiles, publication.year, doi, current_journals, identifiers),
-                          *quartile_labels(imported_quartiles, publication.year, doi, current_journals, identifiers)])),
+            "quartiles": list(dict.fromkeys(f"{item['system']} {item['quartile']}" for item in quartile_entries)),
+            "quartile_entries": quartile_entries,
             "quartile_issues": issues,
             "journals": journals, "issns": [raw[key] for key in ("issn", "eissn", "ISSN", "EISSN") if isinstance(raw.get(key), str)],
             "source_hashes": [publication.source_hash],
@@ -192,14 +194,18 @@ def _unique_articles(articles: list[dict], start: date, end: date, corrections=N
         year = article["year"] = years[0] if len(years) == 1 else None
         article["source_years"] = years
         article["automatic_quartile_bases"] = []
-        if len(article["journals"]) == 1 and not article["quartile_issues"]:
+        if len(article["journals"]) == 1 and not article["quartile_issues"] and not article["journal_corrected"]:
             for item in evidence:
                 if item["year"] != year:
                     continue
-                for label in item["quartiles"]:
-                    if label.endswith((" Q1", " Q2")):
-                        article["automatic_quartile_bases"].append({"basis": label[-2:], "system": label[:-3],
-                            "applicable_year": year, "source": " / ".join(item["origins"]), "source_url": item.get("url")})
+                details = item.get("quartile_entries")
+                if details is None:
+                    # Prepared CSV evidence has no category or metric metadata.
+                    details = [{"quartile": label[-2:], "system": label[:-3],
+                                "source": " / ".join(item["origins"]), "source_url": item.get("url")}
+                               for label in item["quartiles"] if label.endswith((" Q1", " Q2"))]
+                article["automatic_quartile_bases"].extend(
+                    {**entry, "basis": entry["quartile"], "applicable_year": year} for entry in details)
         article["criteria"] = [*article["quartiles"], *article["quartile_issues"]]
         matches = list(dict.fromkeys(k1_journals[key] for name in article["journals"]
                                     if (key := _journal_key(name)) in k1_journals))

app/services/council_assessment.py

@@ -183,7 +183,7 @@ def assess_article(article, proofs, start, end, level, branch, specialty, policy
     basis_conflict = False
     for basis in bases:
         key = ("ВАК", basis_year) if basis["basis"] in {"К1", "К2"} else (
-            basis.get("system", basis["basis"]), basis.get("category"), basis.get("metric_year"), basis_year)
+            basis.get("system", basis["basis"]), basis.get("category"), basis.get("metric"), basis.get("metric_year"), basis_year)
         if key in claims and claims[key] != basis["basis"]:
             reasons.append("Подтверждённые основания источников расходятся")
             basis_conflict = True

app/services/publication_quartiles.py

@@ -129,13 +129,18 @@ def identity_issues(payload, doi="", journals=(), identifiers=()):
                              if (issue := binding_issue(item, payload, doi, journals, identifiers))))
 
 
+def qualifying_entries(payload, publication_year, doi="", journals=(), identifiers=()):
+    return [item for item in entries(payload)
+            if item["quartile"] in {"Q1", "Q2"}
+            and item["system"] in SYSTEM_FIELDS
+            and item.get("publication_year") == publication_year
+            and item.get("metric_year") in {None, publication_year}
+            and not binding_issue(item, payload, doi, journals, identifiers)]
+
+
 def labels(payload, publication_year, doi="", journals=(), identifiers=()):
-    return list(dict.fromkeys(f"{item['system']} {item['quartile']}" for item in entries(payload)
-                             if item["quartile"] in {"Q1", "Q2"}
-                             and item["system"] in SYSTEM_FIELDS
-                             and item.get("publication_year") == publication_year
-                             and item.get("metric_year") in {None, publication_year}
-                             and not binding_issue(item, payload, doi, journals, identifiers)))
+    return list(dict.fromkeys(f"{item['system']} {item['quartile']}" for item in
+                             qualifying_entries(payload, publication_year, doi, journals, identifiers)))
 
 
 def index_sjr(rankings):

app/version.py

@@ -1,3 +1,3 @@
-APP_VERSION = "1.0.0"
-FRONTEND_VERSION = "1.0.0"
-BACKEND_VERSION = "1.0.0"
+APP_VERSION = "1.0.1"
+FRONTEND_VERSION = "1.0.1"
+BACKEND_VERSION = "1.0.1"

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "miem-workers"
-version = "1.0.0"
+version = "1.0.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"] == "1.0.0"
+    assert response.json()["backend_version"] == "1.0.1"
 
 
 def test_api_employees_and_stats_require_admin_session():

tests/test_council.py

@@ -219,7 +219,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 1.0.0 · Frontend 1.0.0" in response.text
+        assert "Backend 1.0.1 · Frontend 1.0.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

@@ -453,3 +453,29 @@ def test_branch_selection_threshold_and_export(db_session, proofs):
             assert parameters['Конец периода'] == datetime(2026, 12, 31)
     finally:
         app.dependency_overrides.clear()
+
+
+@pytest.mark.parametrize('category,metric,expected', [
+    ('Applied Mathematics', 'CiteScore', True),
+    ('Analysis', 'CiteScore', False),
+    ('Analysis', 'SJR', True),
+])
+def test_automatic_quartile_categories_and_metrics_preserve_conflicts(db_session, proofs, category, metric, expected):
+    employee = person(db_session, ['Доктор технических наук'])
+    first = {'system': 'Scopus', 'quartile': 'Q1', 'category': 'Analysis', 'metric': 'CiteScore',
+             'metric_year': 2024, 'publication_year': 2024, 'doi': '10.1234/categories',
+             'journal': 'Journal of Mathematics', 'source': 'Journal metrics 2024',
+             'source_url': 'https://example.org/metrics', 'checked_on': '2026-10-08'}
+    second = {**first, 'quartile': 'Q2', 'category': category, 'metric': metric}
+    db_session.add(EmployeePublication(employee_id=employee.id, title='Categorized article', year=2024,
+        publication_type='ARTICLE', source_hash='categories', doi_url='https://doi.org/10.1234/categories',
+        raw_data={'journal': first['journal']}, quartiles={'schema_version': 1, 'entries': [first, second]}))
+    candidate = result(db_session)
+    assert candidate['q1_count'] == candidate['q2_count'] == candidate['total_count'] == 1
+    assert candidate['confirmed_count'] == int(expected)
+    article = candidate['articles'][0]
+    assert ('Подтверждённые основания источников расходятся' in article['confirmation_reasons']) == (not expected)
+    assert [(basis.get('category'), basis.get('metric'), basis.get('metric_year'), basis['source'], basis['source_url'])
+            for basis in article['automatic_quartile_bases']] == [
+                ('Analysis', 'CiteScore', 2024, first['source'], first['source_url']),
+                (category, metric, 2024, second['source'], second['source_url'])]