fix(scrapers): disambiguate EMISS yearly vs Q1 in dedup key (#1606)

_YEAR_PERIODS ('год') → month=1 collided with Q1 ('I квартал') → month=1
for the same region/year: both produced obs_date=YYYY-01-01, so the second
series in the XML silently overwrote the first in the in-memory by_key dict.

Fix: add _emiss_period_granularity() and extend the dedup key from
(region, obs_date) to (region, obs_date, granularity). 'year' and 'quarter'
are now distinct slots, so both rows survive the parser and reach the upsert.

Test: test_yearly_and_q1_both_survive_dedup verifies len==2 with both
values present; test_period_granularity covers all classification branches.

backend/app/services/scrapers/rosstat_emiss.py

@@ -298,9 +298,7 @@ def _parse_value(raw: str) -> Decimal | None:
     raw = (raw or "").strip()
     if not raw or raw in {"-", "—", "…", "..."}:
         return None
-    cleaned = (
-        raw.replace(" ", "").replace(" ", "").replace(" ", "").replace(",", ".")
-    )
+    cleaned = raw.replace(" ", "").replace(" ", "").replace(" ", "").replace(",", ".")
     try:
         return Decimal(cleaned)
     except (InvalidOperation, ValueError):
@@ -399,9 +397,7 @@ def parse_dataset_rows(
 # ── HTTP fetch (thin) ───────────────────────────────────────────────────────────
 
 
-def _http_get(
-    client: httpx.Client, url: str, *, retries: int = DEFAULT_RETRIES
-) -> bytes:
+def _http_get(client: httpx.Client, url: str, *, retries: int = DEFAULT_RETRIES) -> bytes:
     """GET с ретраями (сеть / 5xx) → байты тела. Module-level для monkeypatch в тестах.
 
     Raises:
@@ -622,9 +618,7 @@ _MONTH_NAME_TO_NUM: dict[str, int] = {
     "декабрь": 12,
 }
 # Годовой под-период (некоторые индикаторы): «значение показателя за год», «год».
-_YEAR_PERIODS: frozenset[str] = frozenset(
-    {"год", "значение показателя за год", "за год"}
-)
+_YEAR_PERIODS: frozenset[str] = frozenset({"год", "значение показателя за год", "за год"})
 
 
 @dataclass(frozen=True)
@@ -704,6 +698,25 @@ def _emiss_period_to_month(period: str) -> int | None:
     return None
 
 
+def _emiss_period_granularity(period: str) -> str:
+    """PERIOD-строку fedstat → гранулярность ('year' | 'quarter' | 'month').
+
+    Используется как третья компонента ключа дедупликации в parse_emiss_sdmx, чтобы
+    годовой ряд ('год') и квартальный Q1 ('I квартал') не перезаписывали друг друга при
+    одинаковых (region, obs_date=YYYY-01-01) (#1606).
+    Возвращает 'unknown' для нераспознанных под-периодов (на практике такие ряды
+    пропускаются ещё до этого вызова через _emiss_period_to_month → None).
+    """
+    p = (period or "").strip().lower()
+    if p in _YEAR_PERIODS:
+        return "year"
+    if p in _QUARTER_START_MONTH:
+        return "quarter"
+    if p in _MONTH_NAME_TO_NUM:
+        return "month"
+    return "unknown"
+
+
 def _series_concepts(series: ET.Element) -> tuple[dict[str, str], dict[str, str]]:
     """Извлечь (SeriesKey, Attributes) concept→value словари из <generic:Series>.
 
@@ -736,7 +749,10 @@ def parse_emiss_sdmx(raw: bytes | str, spec: EmissIndicatorSpec) -> list[EmissRo
          не среди spec.targets — ряд пропускается. PERIOD (Attributes) → стартовый месяц
          под-периода; год — из <Obs><Time>; obs_date = date(year, month, 1). Значение —
          из <Obs><ObsValue value=...> через _parse_value (запятая-десятичный, разделители).
-      3. Дубли по (region, obs_date) схлопываются (последний выигрывает) — PK всё равно один.
+      3. Дубли по (region, obs_date, granularity) схлопываются (последний выигрывает).
+         Ключ трёхкомпонентный — гранулярность ('year'/'quarter'/'month') предотвращает
+         перезапись годового ряда ('год' → obs_date YYYY-01-01) Q1-рядом ('I квартал' →
+         тот же YYYY-01-01) при совместном их присутствии в одном SDMX-ответе (#1606).
 
     Никогда не бросает на данных: битые ячейки/периоды/годы → наблюдение пропускается.
     """
@@ -759,7 +775,10 @@ def parse_emiss_sdmx(raw: bytes | str, spec: EmissIndicatorSpec) -> list[EmissRo
         return []
 
     target_by_okato = {t.oktmo: t for t in spec.targets}
-    by_key: dict[tuple[str, date], EmissRow] = {}
+    # Ключ: (region, obs_date, granularity) — три компоненты предотвращают коллизию
+    # годового ряда ('год' → YYYY-01-01, granularity='year') и Q1 ('I квартал' →
+    # тот же YYYY-01-01, granularity='quarter') при их совместном присутствии в SDMX (#1606).
+    by_key: dict[tuple[str, date, str], EmissRow] = {}
 
     for series in dataset.findall("g:Series", _SDMX_NS):
         key, attrs = _series_concepts(series)
@@ -767,15 +786,17 @@ def parse_emiss_sdmx(raw: bytes | str, spec: EmissIndicatorSpec) -> list[EmissRo
         target = target_by_okato.get(okato)
         if target is None:
             continue
-        month = _emiss_period_to_month(attrs.get("PERIOD", ""))
+        period_raw = attrs.get("PERIOD", "")
+        month = _emiss_period_to_month(period_raw)
         if month is None:
             logger.warning(
                 "EMISS %s: неизвестный PERIOD %r (ОКАТО %s) — пропуск",
                 spec.indicator_id,
-                attrs.get("PERIOD"),
+                period_raw,
                 okato,
             )
             continue
+        granularity = _emiss_period_granularity(period_raw)
         for obs in series.findall("g:Obs", _SDMX_NS):
             time_el = obs.find("g:Time", _SDMX_NS)
             val_el = obs.find("g:ObsValue", _SDMX_NS)
@@ -788,7 +809,7 @@ def parse_emiss_sdmx(raw: bytes | str, spec: EmissIndicatorSpec) -> list[EmissRo
             if value is None:
                 continue
             obs_date = date(int(yr), month, 1)
-            by_key[(target.region, obs_date)] = EmissRow(
+            by_key[(target.region, obs_date, granularity)] = EmissRow(
                 indicator_type=spec.indicator_type,
                 region=target.region,
                 obs_date=obs_date,
@@ -948,9 +969,7 @@ def _parse_year_cell(raw: Any) -> int | None:
     return None
 
 
-def parse_construction_index_xlsx(
-    content: bytes, spec: XlsxIndicatorSpec
-) -> list[MacroRow]:
+def parse_construction_index_xlsx(content: bytes, spec: XlsxIndicatorSpec) -> list[MacroRow]:
     """Pure-парсер открытого xlsx Росстата (индекс цен на СМР) → список MacroRow.
 
     Алгоритм (форма листа — см. блок выше):

backend/tests/scrapers/test_emiss_sdmx.py

@@ -23,6 +23,7 @@ from app.services.scrapers.rosstat_emiss import (
     OKATO_SVERDLOVSK_OBLAST,
     EmissIndicatorSpec,
     RegionTarget,
+    _emiss_period_granularity,
     _emiss_period_to_month,
     parse_emiss_sdmx,
 )
@@ -59,6 +60,56 @@ def test_period_year_and_unknown() -> None:
     assert _emiss_period_to_month("") is None
 
 
+def test_period_granularity() -> None:
+    """_emiss_period_granularity корректно классифицирует все под-периоды (#1606)."""
+    assert _emiss_period_granularity("год") == "year"
+    assert _emiss_period_granularity("значение показателя за год") == "year"
+    assert _emiss_period_granularity("за год") == "year"
+    assert _emiss_period_granularity("I квартал") == "quarter"
+    assert _emiss_period_granularity("IV квартал") == "quarter"
+    assert _emiss_period_granularity("январь") == "month"
+    assert _emiss_period_granularity("декабрь") == "month"
+    assert _emiss_period_granularity("3 неделя") == "unknown"
+    assert _emiss_period_granularity("") == "unknown"
+
+
+def test_yearly_and_q1_both_survive_dedup() -> None:
+    """Годовой ряд ('год') и Q1 ('I квартал') за один год — оба остаются (#1606).
+
+    До фикса: оба давали obs_date=YYYY-01-01, ключ (region, obs_date) был один →
+    второй по XML-порядку перезаписывал первого. После фикса: ключ трёхкомпонентный
+    (region, obs_date, granularity) → 'year' и 'quarter' — разные слоты.
+    """
+    xml = (
+        '<GenericData xmlns="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/message"'
+        ' xmlns:generic="http://www.SDMX.org/resources/SDMXML/schemas/v1_0/generic">'
+        "<DataSet>"
+        # годовой ряд за 2023 — значение 48000
+        "<generic:Series><generic:SeriesKey>"
+        '<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
+        '<generic:Attributes><generic:Value concept="PERIOD" value="год"/>'
+        "</generic:Attributes>"
+        "<generic:Obs><generic:Time>2023</generic:Time>"
+        '<generic:ObsValue value="48000"/></generic:Obs></generic:Series>'
+        # Q1 того же 2023 — значение 42000; obs_date тоже 2023-01-01
+        "<generic:Series><generic:SeriesKey>"
+        '<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
+        '<generic:Attributes><generic:Value concept="PERIOD" value="I квартал"/>'
+        "</generic:Attributes>"
+        "<generic:Obs><generic:Time>2023</generic:Time>"
+        '<generic:ObsValue value="42000"/></generic:Obs></generic:Series>'
+        "</DataSet></GenericData>"
+    )
+    rows = parse_emiss_sdmx(xml, INCOME_PER_CAPITA_SPEC)
+    # Оба наблюдения должны присутствовать — разные гранулярности
+    assert len(rows) == 2, f"ожидались 2 строки, получили {len(rows)}: {rows}"
+    values = {r.value for r in rows}
+    assert Decimal("48000") in values, "годовое наблюдение потеряно"
+    assert Decimal("42000") in values, "Q1-наблюдение потеряно"
+    # obs_date у обоих одинаковый (это нормально — коллизия теперь на стороне DB-upsert)
+    assert all(r.obs_date == date(2023, 1, 1) for r in rows)
+
+
 # ── real-fixture extraction (income id=57039) ─────────────────────────────────────
 
 
@@ -154,13 +205,13 @@ def test_unknown_period_series_skipped() -> None:
         '<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
         '<generic:Attributes><generic:Value concept="PERIOD" value="3 неделя"/>'
         "</generic:Attributes>"
-        '<generic:Obs><generic:Time>2024</generic:Time>'
+        "<generic:Obs><generic:Time>2024</generic:Time>"
         '<generic:ObsValue value="999"/></generic:Obs></generic:Series>'
         "<generic:Series><generic:SeriesKey>"
         '<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
         '<generic:Attributes><generic:Value concept="PERIOD" value="I квартал"/>'
         "</generic:Attributes>"
-        '<generic:Obs><generic:Time>2024</generic:Time>'
+        "<generic:Obs><generic:Time>2024</generic:Time>"
         '<generic:ObsValue value="50000"/></generic:Obs></generic:Series>'
         "</DataSet></GenericData>"
     )
@@ -179,7 +230,7 @@ def test_comma_decimal_value_parsed() -> None:
         '<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
         '<generic:Attributes><generic:Value concept="PERIOD" value="январь"/>'
         "</generic:Attributes>"
-        '<generic:Obs><generic:Time>2024</generic:Time>'
+        "<generic:Obs><generic:Time>2024</generic:Time>"
         '<generic:ObsValue value="113,2"/></generic:Obs></generic:Series></DataSet>'
         "</GenericData>"
     )
@@ -199,13 +250,13 @@ def test_dedup_last_wins() -> None:
         '<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
         '<generic:Attributes><generic:Value concept="PERIOD" value="I квартал"/>'
         "</generic:Attributes>"
-        '<generic:Obs><generic:Time>2024</generic:Time>'
+        "<generic:Obs><generic:Time>2024</generic:Time>"
         '<generic:ObsValue value="100"/></generic:Obs></generic:Series>'
         "<generic:Series><generic:SeriesKey>"
         '<generic:Value concept="s_OKATO" value="65000000000"/></generic:SeriesKey>'
         '<generic:Attributes><generic:Value concept="PERIOD" value="I квартал"/>'
         "</generic:Attributes>"
-        '<generic:Obs><generic:Time>2024</generic:Time>'
+        "<generic:Obs><generic:Time>2024</generic:Time>"
         '<generic:ObsValue value="200"/></generic:Obs></generic:Series>'
         "</DataSet></GenericData>"
     )