chore: план исполнения Среза 2c (12 задач + финальное ревью)
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev
parent
239d5bc270
commit
55e538ce9d
328
docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md
Normal file
328
docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md
Normal file
@ -0,0 +1,328 @@
|
||||
# Срез 2c — загрузка ОСВ, цикл маппинга, документы (arm-core) — план исполнения
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Branch:** `feat/14-slice-2c-osv-ingest` (worktree `.wt/slice-2c/arm-core`). **Issue:** RSHB/arm-core#14. **PR:** #15 (открыт, спека).
|
||||
**Соседний worktree:** docs `.wt/slice-2c/docs` (ветка `feat/16-slice-2c-osv-ingest`, issue #16, PR #17) — там уже лежит `model/data-model-v5.15.dbml`; задачи docs-change 1.2 и 2.1 выполняются в Task 1 и Task 12 этого плана. `../fixtures` — симлинк, только чтение (golden-снимки парсера `golden/expected_osv_parse/**`). `../tabular-parser` — симлинк на main, только чтение (для e2e через compose).
|
||||
**Spec:** `openspec/changes/slice-2c-osv-ingest/` — proposal.md, design.md (D-1…D-13, таблица решений №1–13), specs/{primary-files,osv-result-mapping,documents-api,recognition-job-exchange,data-model,audit-trail}/spec.md. Правила: `../docs/context/05-domain-rules.md` §1.3 (строки, нормализация кода счёта), §1.5 (период), §3 п. 4 (резолв контрагента), §9.2–9.4 (статусы). Спека — единственный источник истины при рулингах; план сгенерирован из `tasks.md` change'а (номера в скобках).
|
||||
|
||||
**Goal:** файл ОСВ, загруженный через API ядра, проходит очередь, распознаётся tabular-parser'ом и превращается в `document` + `osv_entry` без участия человека; статусы файла и документа объяснимы по карточке файла.
|
||||
|
||||
**Architecture:** эндпоинт загрузки → `s3_object` + `stored_file` + `recognition_job(queued)` (outbox существующий) → inbox ставит статусы файла по ack/ошибке → отдельный фоновый цикл маппинга читает результат из S3 и одной транзакцией пишет документы и строки, автозамена прежних версий → API чтения и подтверждения. Чистые функции маппинга — `src/domain/osv_mapping.py`.
|
||||
|
||||
**Tech Stack:** Python 3.11, FastAPI, SQLAlchemy 2 async, Alembic, PostgreSQL 16, pydantic v2 (контракт v1.1 в `src/contract`), miniopy-async, pytest-asyncio, httpx.
|
||||
|
||||
## Global Constraints
|
||||
|
||||
- **DBML v5.15** (`../docs/model/data-model-v5.15.dbml`, 43 таблицы, 27 enum): `osv_data_type + NU`, `app_user.email` в нижнем регистре; полей вне DBML нет. `make dbml-check` — критерий Task 1 и регрессия в каждой задаче.
|
||||
- **Статусы файла** меняет только код, проверивший `stored_file.current_recognition_job_id == job.id` (D-5): загрузка `UPLOADED`; inbox ack → `PROCESSING`; inbox `error`, вахтёр `timeout`, публикатор `dlq` → `ERROR`; цикл маппинга → `PROCESSED`/`ERROR`; автозамена → `SUPERSEDED`.
|
||||
- **Inbox не читает S3 и не создаёт документы**; `ResultMapper`/`NoopMapper` удаляются. Маппинг — `services/result_mapping.py`, чтение S3 вне транзакции, одна транзакция на файл, перед вставкой — очистка строк/документов файла (идемпотентность).
|
||||
- **Строки**: все, кроме `group_type = TOTAL` (OQ-01); `data_type` как пришёл (включая `NU`); суммы — `Decimal` из контракта без изменений; `counterparty_*` как в результате; `is_modified = false`; `parent_entry_id` по `parent_row_number`.
|
||||
- **Резолв контрагента** только для `COUNTERPARTY` с `data_type ∈ {BU, SUM}`: ИНН → точное имя → `normalize_alias` по именам ЮЛ и алиасам; > 1 кандидата → `NULL`; без policy. Имя ЮЛ нормализуется той же `normalize_alias`.
|
||||
- **Статус документа**: `RECOGNIZED` только при `job.status = done`, пустых `checks`-нарушениях, `counterparty_check ≠ MISMATCH` и без предупреждений маппинга; иначе `NEEDS_CHECK`. `period_check = UNKNOWN` (Срез 4).
|
||||
- **Права и policy**: `DOCUMENT_EDIT` (загрузка, удаление, подтверждение), `DOCUMENT_VIEW` (чтение); видимость через ЮЛ-владельца (`legal_entities_visible`); чужое → 404 с тем же телом, что несуществующее.
|
||||
- **Сессии** на маршрутах — только `SessionDep`; фоновые циклы — `async_session_maker`/`session_transaction`. `audit_event` — только `AuditService.record`/`RequestAudit.record` (append-only, тест-страж). Ошибки — `DomainError` (`{"detail": {"code", "message"}}`), 413 — `HTTPException` строкой как в импорте.
|
||||
- **Деньги** — `Decimal`/`Numeric(18,2)`; `float` в домене запрещён.
|
||||
- **Линтеры**: ruff (120, E,F,I,UP,B,PL — константы вместо магических чисел, `# noqa: PLR0913 - причина`), mypy (strict-флаги для `src.domain.*`, `src.models.*`). Тесты: маркеры `integration`, `fixtures`, новый `e2e` (объявить в `pytest.ini`). Команды — из `api/` (`make` из корня). Compose `arm-ka` поднят (PostgreSQL/NATS/MinIO) — не останавливать.
|
||||
- **Коммиты** — русский, по задаче; трейлеры `Co-Authored-By: Claude <модель исполнителя> <noreply@anthropic.com>` и `Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2`. Не пушить, не создавать PR, не `openspec archive`, `openspec/specs/` не трогать (кроме как через change). Коммиты в `../docs` — в его worktree, те же правила.
|
||||
- Допущения: `# ASSUMPTION(OQ-02)` (нарастающая по Q1), `# ASSUMPTION(OQ-55)` (обмен не аудируется), `# ASSUMPTION(OQ-100)` (ИНН из текста не извлекается), `# ASSUMPTION(OQ-103)` (нечисловой первый сегмент → `account_root = None`), `# ASSUMPTION(OQ-22)` (`expires_at` при удалении).
|
||||
|
||||
## Review Focus
|
||||
|
||||
1. **Два задания у одного файла** (повторная загрузка того же содержимого после `SUPERSEDED`, или будущий replay): статус файла меняет только текущее задание; результат старого задания после смены `current_recognition_job_id` файл не трогает — Task 4 (inbox) и Task 7 (цикл): тест «результат чужого задания игнорируется».
|
||||
2. **Результат `needs_check` при отсутствии документов** (парсер по контракту требует `documents` ≥ 1, но файл может прийти от другого сервиса): цикл маппинга → `ERROR` «нет документов ОСВ», не исключение — Task 7: тест.
|
||||
3. **Строка `COUNTERPARTY` с ИНН, которого нет в справочнике, но имя совпадает с алиасом другого ЮЛ** — ИНН не найден → переход к имени/алиасу по §3 (ИНН не блокирует) — Task 6: тест.
|
||||
4. **Имя файла с кириллицей, пробелами и `..`** в ключе S3 — `safe_filename` + ключ без путей; загрузка файла с пустым именем → `UNSUPPORTED_FILE` — Task 3: тест.
|
||||
5. **Подтверждение документа файла, который только что заменён автозаменой** — гонка «подтвердить» vs «цикл заменил»: `confirm` перечитывает файл под `FOR UPDATE` и отвечает 409 `FILE_SUPERSEDED` — Task 9: тест.
|
||||
|
||||
---
|
||||
|
||||
## Задачи
|
||||
|
||||
### Task 1: [1.1] DBML v5.15, миграция 0003, email в нижнем регистре
|
||||
|
||||
**Files:**
|
||||
- Modify: `api/src/models/enums.py` (`OsvDataType.NU` после `BU`), `api/src/models/__init__.py` (`DBML_VERSION = "5.15"`), `Makefile` (`DBML ?= ../../docs/model/data-model-v5.15.dbml`), `api/scripts/dbml_parity.py` (`DBML_FILE_NAME`), `api/tests/test_migrations.py` (`HEAD = "0003_dbml_v5_15"` + тест отката на шаг), `api/tests/test_enums.py`, `api/tests/test_check_dbml_script.py` (фиктивная `v5.16`), `api/scripts/seed_lib/apply.py` (`_apply_users`: `email.strip().lower()`), `api/src/services/users_admin.py` (`create_user`/`update_user`: хранить `email.strip().lower()`; pre-check по `lower` остаётся), `openspec/specs` не трогать; docstring'и «v5.14» → «v5.15» там, где речь о текущей версии
|
||||
- Create: `api/alembic/versions/0003_dbml_v5_15.py`
|
||||
- Create (worktree docs `.wt/slice-2c/docs`): `model/entities-v5.15.md` (копия v5.14 + «Что изменилось в v5.15»: `NU`, email), правки `context/04-data-model-guide.md`, `repo-templates/arm-core.CLAUDE.md`, workspace `/Users/andrey.epifancev/projects/awp-ca/CLAUDE.md` (вне git) → v5.15; `openspec/changes/slice-2c-osv-ingest/tasks.md` 1.2 → `[x]`; коммит в docs
|
||||
|
||||
**Interfaces:**
|
||||
- Миграция (стиль 0002): `upgrade`: `ALTER TYPE dkb.osv_data_type ADD VALUE IF NOT EXISTS 'NU' AFTER 'BU'`; `UPDATE dkb.app_user SET email = lower(email) WHERE email <> lower(email)`; `downgrade`: пересоздание `osv_data_type` со значениями v5.14 (`RENAME TYPE` → `CREATE TYPE` → `ALTER COLUMN osv_entry.data_type TYPE … USING data_type::text::dkb.osv_data_type` → `DROP TYPE`), email не возвращается.
|
||||
- `OsvDataType`: `BU, NU, CURRENCY, QUANTITY, SUM` (порядок DBML).
|
||||
|
||||
- [ ] **Step 1: Тесты.** `test_enums` ожидает `"5.15"` и 5 значений `osv_data_type`; `test_migrations`: `HEAD`, `test_downgrade_one_step` → после `downgrade 0002_dbml_v5_14` `enum_range(NULL::dkb.osv_data_type)` без `NU`; тест данных: вставить `app_user` с `Email@X.RU` на ревизии 0002 → после `upgrade head` email `email@x.ru`; `test_seed_dev`: email dev-пользователей в нижнем регистре; `test_api_admin`: создание с `Qa1@Example.com` → хранится `qa1@example.com`, `/me` по любому регистру.
|
||||
- [ ] **Step 2:** падает → **Step 3:** реализация + docs-правки → **Step 4:** `make dbml-check` пуст; `make migration-cycle`; `make check`; `pytest -m integration tests/test_seed_dev.py tests/test_api_admin.py -q` → PASS.
|
||||
- [ ] **Step 5: Commit** `feat(model): DBML v5.15 — osv_data_type NU, email в нижнем регистре, миграция 0003` (+ коммит в docs `docs: entities-v5.15, guide, ссылки на DBML v5.15`).
|
||||
|
||||
---
|
||||
|
||||
### Task 2: [1.2] S3 `get_bytes`, фабрики, настройки
|
||||
|
||||
**Files:**
|
||||
- Modify: `api/src/core/clients.py` (`S3Storage.get_bytes`), `api/src/services/directory_import.py` (`ObjectStorage` протокол → вынести в `api/src/core/storage.py:ObjectStorage(Protocol)` с `bucket`, `put_object`, `get_bytes`), `api/tests/conftest.py` (`FakeStorage.get_bytes`, `KeyError` → `ObjectNotFoundError`), `api/tests/factories.py`, `api/src/core/config.py` (`UPLOAD_MAX_BYTES: PositiveInt = 50 * 1024 * 1024`, `MAPPING_POLL_SECONDS: PositiveInt = 5`, `MAPPING_MAX_ATTEMPTS: PositiveInt = 3`), `api/.env.example`, `api/.env.test`, `api/pytest.ini` (маркер `e2e`)
|
||||
- Test: `api/tests/test_clients.py` (unit на `get_bytes` с заглушкой Minio), `api/tests/test_factories.py`
|
||||
|
||||
**Interfaces:**
|
||||
```python
|
||||
# core/storage.py
|
||||
class ObjectNotFoundError(Exception): ...
|
||||
class ObjectStorage(Protocol):
|
||||
bucket: str
|
||||
async def put_object(self, obj_name: str, file_data: bytes) -> None: ...
|
||||
async def get_bytes(self, obj_name: str) -> bytes: ... # отсутствие объекта → ObjectNotFoundError
|
||||
# S3Storage.get_bytes: response = await self.client.get_object(self.bucket, obj_name); try: return await response.read() finally: response.close(); await response.release()
|
||||
# (miniopy-async: NoSuchKey → ObjectNotFoundError)
|
||||
# tests/factories.py
|
||||
async def make_stored_file(session, *, entity: LegalEntity, status=FileStatus.UPLOADED, file_type=FileType.XLS, original_name="osv.xls",
|
||||
content_hash: str | None = None, job_status=RecognitionStatus.queued, s3_key: str | None = None,
|
||||
uploaded_by: uuid.UUID | None = None) -> tuple[StoredFile, RecognitionJob] # s3_object + stored_file + job + current_recognition_job_id, flush
|
||||
async def make_document(session, *, file: StoredFile, job: RecognitionJob, account="60", year=2025, quarter=PeriodQuarter.Q1,
|
||||
status=DocumentStatus.RECOGNIZED, is_cumulative=True) -> Document
|
||||
async def make_result_object(session, *, job: RecognitionJob, storage: FakeStorage, result: dict) -> S3Object # кладёт JSON в storage под results/{file_id}/{job_id}/result.json, регистрирует s3_object, job.result_s3_object_id
|
||||
# CommittedJobs.cleanup(): перед stored_file удалять osv_entry, document (по file_id)
|
||||
```
|
||||
|
||||
- [ ] **Step 1:** тесты: `get_bytes` возвращает байты и закрывает response (заглушка с флагами), NoSuchKey → `ObjectNotFoundError`; фабрики создают цепочку и `cleanup` удаляет документы.
|
||||
- [ ] **Step 2–4:** падает → реализация → `make check`, `pytest -m integration tests/test_factories.py -q`.
|
||||
- [ ] **Step 5: Commit** `chore: S3 get_bytes, протокол ObjectStorage, фабрики файлов/документов, настройки загрузки и маппинга`.
|
||||
|
||||
---
|
||||
|
||||
### Task 3: [2.1] Загрузка файлов
|
||||
|
||||
**Files:**
|
||||
- Create: `api/src/domain/file_types.py`, `api/src/services/primary_files.py`, `api/src/schemas/files.py`, `api/src/api/endpoints/files/__init__.py`, `api/src/api/endpoints/files/upload.py`, `api/src/api/dependencies/files.py`
|
||||
- Modify: `api/src/api/router.py` (include `files.router`), `api/src/services/directory_import.py` (`safe_filename`, `CONTENT_TYPES` → вынести в `api/src/common/filenames.py` и переиспользовать)
|
||||
- Test: `api/tests/test_domain_file_types.py`, `api/tests/test_api_files_upload.py`
|
||||
|
||||
**Interfaces:**
|
||||
```python
|
||||
# domain/file_types.py
|
||||
@dataclass(frozen=True)
|
||||
class DetectedFile: file_type: FileType; pipeline: RecognitionPipeline; content_type: str
|
||||
def detect_file(name: str, content: bytes) -> DetectedFile # расширение → тип; сигнатура: xls = b"\xd0\xcf\x11\xe0", xlsx/xlsb = b"PK\x03\x04", pdf = b"%PDF", png = b"\x89PNG", jpg = b"\xff\xd8\xff"; несовпадение/неизвестно → DomainValidationError("UNSUPPORTED_FILE", …)
|
||||
MAX_ATTEMPTS = {RecognitionPipeline.tabular: 3, RecognitionPipeline.image: 1}
|
||||
# services/primary_files.py
|
||||
@dataclass(frozen=True)
|
||||
class AcceptedFile: file_id: uuid.UUID; job_id: uuid.UUID; name: str
|
||||
@dataclass(frozen=True)
|
||||
class RejectedFile: name: str; code: str; message: str; file_id: uuid.UUID | None = None
|
||||
class PrimaryFilesService:
|
||||
def __init__(self, session, audit: RequestAudit, scope: VisibilityScope, storage: ObjectStorage | None, settings: Settings) -> None
|
||||
async def upload_one(self, entity: LegalEntity, name: str, content: bytes) -> AcceptedFile # raises DomainError → эндпоинт превращает в RejectedFile; каждая загрузка — begin_nested() (SAVEPOINT) внутри транзакции запроса, чтобы отказ одного файла не откатил принятые
|
||||
async def find_duplicate(self, entity_id, sha256) -> StoredFile | None # status != SUPERSEDED
|
||||
# endpoint: POST /legal-entities/{legal_entity_id}/files, files: list[UploadFile] = File(...), require("DOCUMENT_EDIT"); ЮЛ через get_legal_entity(session, scope, id) → 404; storage None → HTTPException(503, "хранилище файлов недоступно") (код STORAGE_UNAVAILABLE в detail-строке); размер — _read_limited(file, settings.UPLOAD_MAX_BYTES) → 413 превращается в RejectedFile(FILE_TOO_LARGE) (не HTTPException на весь запрос); ответ UploadResult(accepted, rejected), 201 если accepted непуст, иначе 422
|
||||
# ключ S3: f"uploads/{file_id}/{job_id}/{safe_filename(name)}"; S3Object (bucket_role=uploads, sha256, size_bytes, content_type, created_by=f"user:{id}", created_at) → flush → put_object → StoredFile(status=UPLOADED, version_no=1, content_hash=sha256, original_name=safe name, uploaded_by, uploaded_at) → RecognitionJob(file_id, pipeline, request_subject=request_subject(pipeline) из contract.subjects, schema_version=CONTRACT_VERSION, status=queued, max_attempts) → flush → file.current_recognition_job_id = job.id; audit FILE_UPLOAD (object_id=file.id, after_value={original_name, content_hash, job_id, file_type})
|
||||
```
|
||||
- `schemas/files.py`: `UploadResult{accepted: list[AcceptedOut], rejected: list[RejectedOut]}`, `JobBrief{id, status, error_code, error_message, warnings: list[str], mapping_warnings: list[str], finished_at}`, `DocumentBrief{id, document_type, osv_account_code, period_year, period_quarter, is_cumulative, status, period_check, counterparty_check, locator}`, `FileOut{id, legal_entity_id, original_name, file_type, status, version_no, size_bytes, content_hash, uploaded_at, uploaded_by, superseded_by_file_id, job: JobBrief | None, documents: list[DocumentBrief]}` (используется Task 9).
|
||||
|
||||
- [ ] **Step 1: Тесты.** `test_domain_file_types.py`: 6 типов по сигнатуре, `osv.xlsx` с BIFF8 → `UNSUPPORTED_FILE`, `.docx` → `UNSUPPORTED_FILE`, пустое имя → `UNSUPPORTED_FILE` (Review Focus 4). `test_api_files_upload.py` (integration, `auth_client`, `app.state.storage = FakeStorage()`): сценарии спеки `primary-files` «Реальная ОСВ принята» (байты `../fixtures/osv/rb/2024-FY/ОСВ_66 за 2024.xls`, маркер `fixtures`; затем `publish_pending` с `FakeNats` из `tests/test_recognition_exchange.py` публикует запрос с ключом `uploads/{file_id}/{job_id}/ОСВ_66 за 2024.xls`), «Смесь» (xlsx + docx + 51 МиБ через `UPLOAD_MAX_BYTES=10` в настройках теста), «Сигнатура», «Чужое ЮЛ» (`client_manager` → 404), «Дубль» (второй раз → `DUPLICATE_FILE` с `file_id`; тот же файл у другого ЮЛ → принят); имя `../..\\ОСВ 60 (2).xls` → ключ без `..` и слэшей; `storage=None` → 503; два `FILE_UPLOAD` в аудите.
|
||||
- [ ] **Step 2–4:** падает → реализация → `pytest -m "integration or fixtures" tests/test_api_files_upload.py -q`, `make check`.
|
||||
- [ ] **Step 5: Commit** `feat(files): загрузка файлов организации — S3, stored_file, recognition_job в outbox, дедуп, аудит`.
|
||||
|
||||
---
|
||||
|
||||
### Task 4: [2.2] Статусы файла в обмене, удаление `ResultMapper`
|
||||
|
||||
**Files:**
|
||||
- Modify: `api/src/services/recognition_exchange.py` (`_apply_ack` → `PROCESSING`; `_apply_result` при `error` → `ERROR`; `expire_deadlines` — второй `UPDATE stored_file`; `_process_one` при `dlq` → `ERROR`; удалить `ResultMapper`, `NoopMapper`, `_map_result`, `_MapperError`, параметр `mapper` у `handle_message`), `api/src/app.py` (`partial(handle_message, session_factory=...)`), `api/tests/test_recognition_inbox.py`, `api/tests/test_recognition_exchange.py` (`RecordingMapper` → проверки статуса файла)
|
||||
- Create: `api/src/services/file_status.py`
|
||||
|
||||
**Interfaces:**
|
||||
```python
|
||||
# services/file_status.py
|
||||
async def set_file_status_for_job(session, job: RecognitionJob, status: FileStatus, *, now: datetime) -> bool
|
||||
# UPDATE stored_file SET status=:status, updated_at=:now WHERE id = job.file_id AND current_recognition_job_id = job.id AND status NOT IN (SUPERSEDED, PROCESSED); returns rowcount == 1
|
||||
# в expire_deadlines: после UPDATE recognition_job … RETURNING id, file_id → UPDATE stored_file SET status='ERROR' WHERE current_recognition_job_id = ANY(ids) AND status NOT IN (SUPERSEDED, PROCESSED)
|
||||
```
|
||||
Ошибка задания (`error`/`timeout`/`dlq`) ставит файлу `ERROR`; ack — `PROCESSING` только из `UPLOADED`; `done`/`needs_check` файл не трогают.
|
||||
|
||||
- [ ] **Step 1: Тесты** (integration, `CommittedJobs` + `make_stored_file`): ack → `PROCESSING`; `error` → `ERROR`; `done` → файл `PROCESSING`; `timeout` вахтёром → `ERROR`; `dlq` публикатором → `ERROR`; результат задания, которое уже не `current_recognition_job_id` файла → файл не меняется (Review Focus 1); существующие тесты inbox без `mapper` зелёные.
|
||||
- [ ] **Step 2–4:** падает → реализация → `pytest -m integration tests/test_recognition_inbox.py tests/test_recognition_exchange.py -q`, `make check`.
|
||||
- [ ] **Step 5: Commit** `feat(exchange): статусы stored_file по ack/ошибке/таймауту/dlq; inbox без ResultMapper (OQ-67)`.
|
||||
|
||||
---
|
||||
|
||||
### Task 5: [3.1a] Домен маппинга — строки, нормализация, проверки
|
||||
|
||||
**Files:**
|
||||
- Create: `api/src/domain/osv_mapping.py`
|
||||
- Test: `api/tests/test_domain_osv_mapping.py`
|
||||
|
||||
**Interfaces:**
|
||||
```python
|
||||
ACCOUNT_SEGMENT_DIGITS = 2
|
||||
def normalize_account_code(raw: str) -> str # "60.1"→"60.01", "76.7.1"→"76.07.1", "58.03"→"58.03", "76.АВ"→"76.АВ", "МЦ.04"→"МЦ.04", "001"→"001"; сегменты по '.', числовой сегмент (1–3 цифры) → zfill(2) кроме первого (как есть) и третьего (как есть)
|
||||
def account_root(code: str) -> str | None # первый сегмент, если он из цифр; иначе None (ASSUMPTION(OQ-103))
|
||||
@dataclass(frozen=True)
|
||||
class EntryDraft: id: uuid.UUID; row_number: int; level: int; parent_id: uuid.UUID | None; row_value: str; group_type: OsvGroupType; group_name: str | None; data_type: OsvDataType | None; account_code: str | None; account_root: str | None; account_id: uuid.UUID | None; counterparty_name: str | None; counterparty_inn: str | None; sums: tuple[Decimal | None, ...] # opening_debit, opening_credit, turnover_debit, turnover_credit, closing_debit, closing_credit
|
||||
@dataclass(frozen=True)
|
||||
class MappedSheet: entries: list[EntryDraft]; root: EntryDraft | None; total_present: bool; orphans: tuple[int, ...]
|
||||
def build_entries(doc: OsvDocument, chart: Mapping[str, uuid.UUID]) -> MappedSheet
|
||||
# пропуск TOTAL; id = uuid4 заранее; parent по parent_row_number (нет в карте → parent None; MappedSheet.orphans — список row_number, вызывающий пишет предупреждение); account_code = row.account_code_raw; account_id = chart.get(normalize_account_code(account_code)) if account_code else None; group_type TOTAL не встречается в entries; data_type через OsvDataType(row.data_type) if row.data_type else None; counterparty_* как есть
|
||||
def document_fields(doc: OsvDocument) -> DocumentDraft # osv_account_code = normalize_account_code(header.account_code_raw), period_year, period_quarter (PeriodQuarter), is_cumulative, locator
|
||||
def counterparty_check(header: OsvHeader, owner_inn: str, owner_names_normalized: frozenset[str]) -> DocumentCheckResult
|
||||
# organization_inn есть → MATCH если == owner_inn (только цифры), иначе MISMATCH; нет ИНН: normalize_alias(organization_name) ∈ owner_names_normalized → MATCH; пустое имя и ИНН → UNKNOWN; иначе MISMATCH
|
||||
def document_status(*, job_done: bool, checks: OsvChecks | None, counterparty: DocumentCheckResult, warnings: Sequence[str]) -> DocumentStatus
|
||||
# RECOGNIZED iff job_done and (checks is None or (checks.total_row_matches_root is not False and not checks.balance_identity_violations and not checks.parent_sum_violations)) and counterparty != MISMATCH and not warnings; else NEEDS_CHECK
|
||||
def opening_matches(a: EntryDraft, b_opening: tuple[Decimal | None, Decimal | None]) -> bool # ±0.01, None = 0 — для D-9
|
||||
```
|
||||
|
||||
- [ ] **Step 1: Тесты** (unit): нормализация (таблица из 8 случаев), `account_root` (`60`, `001`, `МЦ.04` → None); `build_entries` на `api/contracts/examples/result-osv-minimal.json` (TOTAL отсутствует, parent выставлен, суммы `Decimal`) и на golden `../fixtures/golden/expected_osv_parse/rb/2024-FY/ОСВ_66 за 2024.json` (маркер `fixtures`; `chart` — словарь из seed CSV): 20 строк, у «Юг-Финанс / ПЕКАРНЯ СЧАСТЛИВОЧНАЯ» `account_code = "66.03"`, `account_id` задан; `counterparty_check` — ИНН совпал/не совпал, по имени, по алиасу, пустая шапка; `document_status` — 5 комбинаций; `opening_matches` на границе 0.01.
|
||||
- [ ] **Step 2–4:** падает → реализация → `pytest tests/test_domain_osv_mapping.py -q`, `mypy src` strict для `src.domain.*`.
|
||||
- [ ] **Step 5: Commit** `feat(domain): маппинг osvDocument в черновики документа и строк, нормализация кода счёта, проверки (§1.3, §9.3)`.
|
||||
|
||||
---
|
||||
|
||||
### Task 6: [3.1b] Резолв контрагента
|
||||
|
||||
**Files:**
|
||||
- Create: `api/src/repositories/counterparties.py`
|
||||
- Test: `api/tests/test_counterparty_resolve.py`
|
||||
|
||||
**Interfaces:**
|
||||
```python
|
||||
@dataclass
|
||||
class CounterpartyIndex:
|
||||
by_inn: dict[str, uuid.UUID]
|
||||
by_exact_name: dict[str, set[uuid.UUID]] # strip()
|
||||
by_normalized: dict[str, set[uuid.UUID]] # normalize_alias(name) ∪ normalized_alias алиасов
|
||||
@classmethod
|
||||
async def load(cls, session) -> "CounterpartyIndex" # два запроса: legal_entity(id, inn, name), legal_entity_alias(legal_entity_id, normalized_alias); без policy
|
||||
def resolve(self, inn: str | None, name: str | None) -> uuid.UUID | None
|
||||
# inn (цифры) in by_inn → id; else name.strip() in by_exact_name и ровно один → id (> 1 → None, 0 → дальше); else normalize_alias(name) in by_normalized и ровно один → id; иначе None
|
||||
def owner_names(self, entity: LegalEntity) -> frozenset[str] # normalize_alias(name) ∪ его алиасы — для counterparty_check
|
||||
```
|
||||
|
||||
- [ ] **Step 1: Тесты** (integration на `db_session` с фабриками `make_legal_entity` + алиасы через `LegalEntityAlias`): сценарий спеки «По ИНН, по алиасу, неоднозначно»; ИНН не найден, имя совпадает с алиасом другого ЮЛ → по алиасу (Review Focus 3); закрытое ЮЛ (`close_date`) тоже резолвится; `owner_names`.
|
||||
- [ ] **Step 2–4:** падает → реализация → тесты, `make check`.
|
||||
- [ ] **Step 5: Commit** `feat(mapping): индекс и резолв контрагентов по ИНН/имени/алиасу (§3 п. 4)`.
|
||||
|
||||
---
|
||||
|
||||
### Task 7: [3.2] Цикл маппинга
|
||||
|
||||
**Files:**
|
||||
- Create: `api/src/services/result_mapping.py`
|
||||
- Modify: `api/src/services/recognition_exchange.py` (`ExchangeTasks.start` принимает `storage` и добавляет задачу `run_mapping_loop`), `api/src/app.py` (`_start_exchange(nats, storage)`; цикл маппинга стартует вместе с обменом, при `NATS_ENABLED=false` не стартует)
|
||||
- Test: `api/tests/test_result_mapping.py`
|
||||
|
||||
**Interfaces:**
|
||||
```python
|
||||
MAPPING_ATTEMPTS_KEY = "mapping_attempts"; MAPPING_ERROR_KEY = "mapping_error"; MAPPING_WARNINGS_KEY = "mapping_warnings"
|
||||
class MappingFatal(Exception): """ошибка без повтора (ключ, объект, схема, нет документов)"""
|
||||
@dataclass(frozen=True)
|
||||
class Candidate: job_id: uuid.UUID; file_id: uuid.UUID; bucket: str; key: str
|
||||
async def pick_candidate(session) -> Candidate | None
|
||||
# SELECT job.id, job.file_id, s3.bucket, s3.key FROM recognition_job job JOIN stored_file f ON f.id=job.file_id AND f.current_recognition_job_id=job.id JOIN s3_object s3 ON s3.id=job.result_s3_object_id
|
||||
# WHERE job.status IN (done, needs_check) AND f.status='PROCESSING' AND coalesce((job.metrics->>'mapping_attempts')::int,0) < :max ORDER BY job.finished_at, job.id LIMIT 1 FOR UPDATE OF job SKIP LOCKED
|
||||
def validate_result_key(candidate: Candidate, storage_bucket: str) -> None # MappingFatal при бакете ≠ storage.bucket или ключе не вида results/{file_id}/{job_id}/…
|
||||
async def map_file(session, *, job: RecognitionJob, file: StoredFile, result: RecognitionResult, chart: Mapping[str, uuid], index: CounterpartyIndex, now: datetime) -> list[str] # предупреждения
|
||||
# в транзакции: DELETE osv_entry WHERE document_id IN (документы файла); DELETE document WHERE stored_file_id; по osvDocument: document_fields → если osv_account_code уже встречался в этом файле → warning "sheet=<locator>: счёт <код> уже есть в листе <locator первого>", первый документ помечается needs_check; иначе Document(...); build_entries → insert(OsvEntry).values([...]) пакетами по 1000; D-9: для Q2–Q4 с is_cumulative False — ищем Document того же ЮЛ/счёта/года Q1 с файлом != SUPERSEDED, корень = osv_entry с row_number минимальным level 0 → opening_matches → is_cumulative True + warning; counterparty_check(header, file.legal_entity.inn, index.owner_names(entity)); status = document_status(...); job.group_schema = [{"locator", "groups":[{"type","name","has_inn"}]}]; job.metrics[mapping_warnings] = warnings; file.status = PROCESSED, updated_at; вызов supersede_previous добавляет Task 8; в этой задаче map_file завершается установкой PROCESSED
|
||||
async def process_once(session_factory, storage: ObjectStorage, settings, *, now=None) -> bool
|
||||
# 1) короткая транзакция: pick_candidate → нет → False; validate_result_key (MappingFatal → в той же транзакции файл ERROR, metrics.mapping_error, commit, return True)
|
||||
# 2) без транзакции: content = await storage.get_bytes(key) (ObjectNotFoundError → fatal); RecognitionResult.model_validate_json (ValidationError → fatal); нет osvDocument → fatal "нет документов ОСВ"
|
||||
# 3) транзакция: заново SELECT job FOR UPDATE и file; если file.status != PROCESSING или current_job != job → return True (кто-то опередил); chart = {code: id} одним запросом; index = CounterpartyIndex.load; map_file; commit
|
||||
# исключение в 3): rollback; отдельная транзакция: metrics.mapping_attempts += 1; если >= MAPPING_MAX_ATTEMPTS → file ERROR + mapping_error; лог с traceback
|
||||
async def run_mapping_loop(session_factory, storage, settings, *, sleep=asyncio.sleep) -> None # while True: processed = await process_once(...); await sleep(0 if processed else settings.MAPPING_POLL_SECONDS); исключения цикла логируются, цикл не умирает
|
||||
```
|
||||
`ExchangeTasks.start(session_factory, nats, settings, storage)` — третья задача `recognition-result-mapper` только при `storage is not None` (иначе warning «S3 не подключено: маппинг результатов не запущен»).
|
||||
|
||||
- [ ] **Step 1: Тесты** (integration, `committed_sessionmaker` + `FakeStorage`; результат — golden JSON `ОСВ_66 за 2024.json` с подставленными `job_id`/`file_id`): сценарии спеки `osv-result-mapping` «Результат смапплен» (документ 66/2024/Q4/cumulative, `locator`, 20 строк, `group_schema`), «Ключ вне папки» → `ERROR`, «Сбой с повтором» (monkeypatch `map_file` бросает 2 раза → attempts, 3-й → `ERROR`), «Идемпотентность» (вставить половину строк вручную → `process_once` → ровно один набор), «Два листа одного счёта» (JSON с двумя документами `60`) → один документ `NEEDS_CHECK` + warning, «Шапка другой организации» → `MISMATCH`/`NEEDS_CHECK`, «Нарастающая по данным» (сначала документ Q1 через фабрики с корнем, затем Q3 с теми же opening) → `is_cumulative = True`, «Чистый результат» → `RECOGNIZED`; результат без документов (`status: error`-подобный JSON с `documents: []` не валиден по контракту → fatal) → `ERROR` (Review Focus 2); результат для задания, которое уже не текущее → не трогает файл (Review Focus 1); `run_mapping_loop` с заглушкой sleep обрабатывает два кандидата подряд.
|
||||
- [ ] **Step 2–4:** падает → реализация → `pytest -m "integration or fixtures" tests/test_result_mapping.py -q`, `make check`.
|
||||
- [ ] **Step 5: Commit** `feat(mapping): фоновый цикл маппинга результата в document/osv_entry (OQ-69, OQ-70)`.
|
||||
|
||||
---
|
||||
|
||||
### Task 8: [3.3] Автозамена
|
||||
|
||||
**Files:**
|
||||
- Modify: `api/src/services/result_mapping.py` (`supersede_previous`), `api/src/services/audit.py` — без изменений (системное событие `user_id=None`)
|
||||
- Test: `api/tests/test_result_mapping.py` (+ сценарии)
|
||||
|
||||
**Interfaces:**
|
||||
```python
|
||||
async def supersede_previous(session, *, new_file: StoredFile, new_documents: Sequence[Document], uploader_id: uuid.UUID | None, now) -> list[StoredFile]
|
||||
# для каждого нового документа: SELECT stored_file JOIN document WHERE stored_file.legal_entity_id = new.legal_entity_id AND stored_file.id != new.id AND stored_file.status != SUPERSEDED AND document.(type, osv_account_code, period_year, period_quarter) совпадают; distinct файлы → status SUPERSEDED, superseded_by_file_id = new.id, updated_at; new.version_no = max(v.version_no) + 1 (если нашлись); AuditService(session).record(action=FILE_REPLACE, section=PRIMARY_DOCUMENTS, object_type="stored_file", object_id=old.id, user_id=None, after_value={"file_id": new.id, "version_no": new.version_no}, description=f"заменён файлом {new.original_name} (загрузил user:{uploader_id}); документы: OSV 60 2025 Q1 (CONFIRMED)…; confirmed={true|false}")
|
||||
```
|
||||
Вызывается в `map_file` после установки `PROCESSED`, в той же транзакции.
|
||||
|
||||
- [ ] **Step 1: Тесты:** сценарий спеки «Новая ОСВ 60 за 1 кв. 2025» (старый `CONFIRMED` → `SUPERSEDED`, `version_no = 2`, аудит с `confirmed=true`); два разных счёта не заменяют друг друга; тот же счёт другого года не заменяет; заменённый файл не появляется в `GET files` без `include_superseded` (проверяется в Task 9, здесь — по БД).
|
||||
- [ ] **Step 2–4:** падает → реализация → тесты, `make check`.
|
||||
- [ ] **Step 5: Commit** `feat(mapping): автозамена прежних версий файла после маппинга (FILE_REPLACE)`.
|
||||
|
||||
---
|
||||
|
||||
### Task 9: [4.1] API чтения и подтверждение документа
|
||||
|
||||
**Files:**
|
||||
- Create: `api/src/repositories/files.py`, `api/src/services/documents.py`, `api/src/schemas/documents.py`, `api/src/api/endpoints/files/read.py`, `api/src/api/endpoints/files/documents.py`
|
||||
- Modify: `api/src/api/endpoints/files/__init__.py`, `api/src/api/dependencies/files.py`, `api/tests/test_openapi_routes.py`
|
||||
- Test: `api/tests/test_api_files_read.py`, `api/tests/test_api_documents.py`
|
||||
|
||||
**Interfaces:**
|
||||
```python
|
||||
# repositories/files.py
|
||||
async def list_files(session, scope, entity_id, *, include_superseded: bool, limit, offset) -> tuple[list[StoredFile], int] # с joinedload job (current) и documents (selectin)
|
||||
async def get_file(session, scope, file_id) -> StoredFile | None # через legal_entities_visible по владельцу
|
||||
async def get_document(session, scope, document_id) -> Document | None
|
||||
async def list_entries(session, document_id, *, limit, offset) -> tuple[list[OsvEntry], int] # ORDER BY row_number
|
||||
# services/documents.py
|
||||
class DocumentService:
|
||||
async def confirm(self, document_id) -> Document # SELECT document JOIN stored_file FOR UPDATE; file.status == SUPERSEDED → ConflictError("FILE_SUPERSEDED"); status == CONFIRMED → ConflictError("ALREADY_CONFIRMED"); status ∉ {RECOGNIZED, NEEDS_CHECK} → ConflictError("DOCUMENT_NOT_CONFIRMABLE"); set CONFIRMED, confirmed_by, confirmed_at; audit DOCUMENT_CONFIRM before/after status
|
||||
# schemas/documents.py: DocumentOut{id, file_id, legal_entity_id, document_type, osv_account_code, period_year, period_quarter, is_cumulative, locator, status, period_check, counterparty_check, confirmed_by, confirmed_at, group_schema: list[GroupOut] | None, warnings: list[str]}; EntryOut{id, row_number, level, parent_entry_id, row_value, group_type, group_name, data_type, account_code, account_root, account_id, counterparty_name, counterparty_inn, counterparty_id, opening_debit…closing_credit (Decimal → str)}; EntriesPageParams(limit ≤ 2000, default 500)
|
||||
# маршруты: GET /legal-entities/{id}/files → Page[FileOut]; GET /files/{file_id} → FileOut; GET /documents/{id} → DocumentOut; GET /documents/{id}/entries → Page[EntryOut]; POST /documents/{id}/confirm → DocumentOut
|
||||
```
|
||||
|
||||
- [ ] **Step 1: Тесты:** сценарии спеки `documents-api` (список с `PROCESSED`+документом и `ERROR`, `include_superseded`, карточка, 404 для чужого, строки деревом на данных из golden через `make_entries_from_result`, `limit=2001` → 422, подтверждение с аудитом, повтор → 409, документ `SUPERSEDED` файла → 409; гонка Review Focus 5: файл переведён в `SUPERSEDED` между чтением и `confirm` — имитация прямым UPDATE перед вызовом → 409); `test_openapi_routes` — новые пути.
|
||||
- [ ] **Step 2–4:** падает → реализация → тесты, `make check`.
|
||||
- [ ] **Step 5: Commit** `feat(files): API файлов, документов и строк ОСВ; подтверждение документа`.
|
||||
|
||||
---
|
||||
|
||||
### Task 10: [4.2] Удаление файла
|
||||
|
||||
**Files:**
|
||||
- Modify: `api/src/services/primary_files.py` (`delete`), `api/src/api/endpoints/files/read.py` (`DELETE /files/{file_id}`), `api/src/repositories/files.py`
|
||||
- Test: `api/tests/test_api_files_delete.py`
|
||||
|
||||
**Interfaces:**
|
||||
```python
|
||||
async def delete(self, file_id) -> None # get_file(scope) → 404; есть document CONFIRMED → ConflictError("FILE_HAS_CONFIRMED_DOCUMENTS"); EXISTS stored_file WHERE superseded_by_file_id = id → ConflictError("FILE_SUPERSEDES_OTHERS"); before = FileOut-словарь; DELETE osv_entry (по документам файла), document, recognition_job WHERE file_id (сначала file.current_recognition_job_id = NULL + flush — FK цикл), stored_file; s3_object.expires_at = now (ASSUMPTION(OQ-22)); audit FILE_DELETE before_value=before
|
||||
```
|
||||
|
||||
- [ ] **Step 1: Тесты:** сценарии спеки (ошибочный файл → 204, записей нет, `expires_at`, аудит; `CONFIRMED` → 409; файл, заменивший другой → 409; `SUPERSEDED` → 204; чужой → 404; `PROCESSED` без подтверждения → 204 со строками).
|
||||
- [ ] **Step 2–4:** падает → реализация → тесты, `make check`.
|
||||
- [ ] **Step 5: Commit** `feat(files): удаление файла с ограничениями, expires_at у объекта S3, FILE_DELETE`.
|
||||
|
||||
---
|
||||
|
||||
### Task 11: [5.1] Сквозной прогон через compose
|
||||
|
||||
**Files:**
|
||||
- Create: `api/tests/e2e/__init__.py`, `api/tests/e2e/test_osv_ingest_e2e.py`, `api/tests/e2e/conftest.py`
|
||||
- Modify: `Makefile` (`e2e: cd api && pytest -m e2e`, `check-full` включает `e2e`), `README.md` (как поднять профиль `services`)
|
||||
|
||||
**Interfaces:** тест (маркеры `integration`, `e2e`, `fixtures`): предусловия — `curl localhost:8001/healthcheck` tabular-parser из профиля `services` (`docker compose -f ../docs/dev/docker-compose.yml --profile services up -d --build tabular-parser`) и `NATS_ENABLED=true` у тестового приложения; иначе `pytest.skip("профиль services не поднят")`. Шаги: полное приложение `src.app.app` (lifespan с реальными NATS/MinIO из `.env.test`, `NATS_ENABLED=true` через monkeypatch settings) на отдельной БД (`committed_sessionmaker` + seed) → `POST /legal-entities/{id}/files` для ЮЛ, созданного фабрикой с ИНН `3528052718` (АО «Русский бисквит», владелец ОСВ 66) с `rb/2024-FY/ОСВ_66 за 2024.xls` → ждать до 60 с `GET /files/{id}` → `PROCESSED` → документ 66/2024/Q4, 20 строк, `counterparty_check = MATCH`, `status = RECOGNIZED`; очистка: удалить файл через API.
|
||||
|
||||
- [ ] **Step 1:** тест; **Step 2:** поднять профиль (`make up` + `--profile services` для `tabular-parser`), прогнать `make e2e` → PASS; без профиля → skip; `make check-full` зелёный.
|
||||
- [ ] **Step 3: Commit** `test(e2e): сквозной прогон загрузки ОСВ через compose с реальным tabular-parser`.
|
||||
|
||||
---
|
||||
|
||||
### Task 12: [5.2] Документы и `make check-full`
|
||||
|
||||
**Files:**
|
||||
- Modify: `README.md` (разделы «Первичные документы: загрузка», «Статусы файла и документа», «Цикл маппинга» с настройками, «API файлов и документов», «Удаление и замена», «Ограничения»), `CLAUDE.md` (DBML v5.15, `domain/osv_mapping.py`, `services/primary_files.py`, `services/result_mapping.py`, три фоновых цикла, маркер `e2e`), `openspec/changes/slice-2c-osv-ingest/tasks.md` (все `[x]` кроме 5.3), `openspec/changes/slice-2c-osv-ingest/design.md` (рулинги контроллера)
|
||||
- Modify (worktree docs): `context/05-domain-rules.md` (§1.3, §1.5, §3 п. 4, §9.2–9.3, новый §9.5 по факту), `context/03-architecture.md` (цикл маппинга), `context/07-open-questions.md` (закрыть OQ-67/69/70/59/109; новые OQ: удаление после консолидаций, ретеншн при удалении), `context/09-delivery-plan.md` (2c по факту), `dev/qa-guide.md` (раздел «Загрузка ОСВ и просмотр строк через API»: curl-сценарий, ожидания на `ОСВ_66`), `openspec/changes/slice-2c-osv-ingest/tasks.md` 2.1 → `[x]`
|
||||
|
||||
- [ ] **Step 1:** таблица «требование спеки → сценарий → тест» в отчёте задачи; пропуски — добавить тесты.
|
||||
- [ ] **Step 2:** документы; `openspec validate --strict` в обоих репозиториях.
|
||||
- [ ] **Step 3:** `make check-full` (compose поднят; профиль `services` — по возможности, иначе e2e skip) → зелёный; вывод — в ledger.
|
||||
- [ ] **Step 4: Commit** здесь `docs: README/CLAUDE.md Среза 2c, tasks.md`; в docs `docs: §1.3/§1.5/§3/§9 по факту маппинга, архитектура (цикл маппинга), OQ, план срезов, qa-guide`.
|
||||
|
||||
---
|
||||
|
||||
## Завершение
|
||||
|
||||
Финальное ревью всей ветки (opus): спека ↔ код по шести дельтам; инварианты (поля вне DBML v5.15, `audit_event` append-only, код роли, `float`, первичка не мутируется — `osv_entry` пишется один раз и только маппером); безопасность (policy на всех маршрутах файлов/документов, 404 vs 403, `safe_filename`, ключ результата OQ-70, лимит размера, 503 без S3); гонки (два цикла + inbox на `stored_file`, `confirm` vs автозамена); производительность маппинга (1045 строк `ОСВ_60 за 2024.xls` — цель < 2 с). Находки `C/I/M` со статусами; `deferred` — в PR и follow-up issue при закрытии.
|
||||
Loading…
x
Reference in New Issue
Block a user