16b53b0bdb
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
344 lines
19 KiB
Python
344 lines
19 KiB
Python
"""Очистка форм ББ/ОФР перед выкладкой ядра с миграцией v5.19 (spec arm-core «Очистка форм перед выкладкой»,
|
||
Р-24, Р-27).
|
||
|
||
Формы пересоздаются повторной загрузкой файлов уже по новым правилам. Запуск из `api/`:
|
||
|
||
.venv/bin/python -m scripts.purge_statements # отчёт (--dry-run), база не меняется
|
||
.venv/bin/python -m scripts.purge_statements --apply # очистка, одна транзакция
|
||
.venv/bin/python -m scripts.purge_statements --apply --include-mixed # смешанные файлы удаляются целиком
|
||
|
||
Отбор — документы `BALANCE` / `INCOME` файлов любого статуса.
|
||
Отказ без изменений (P3-12, Р-35) — только если отобранный (или удаляемый вместе с файлом) документ упомянут в
|
||
`ex_snapshot_item` либо привязан к верифицированной версии: `VERIFIED` или `ARCHIVED` со срезом `ex_snapshot`
|
||
(`source_version_id` = версия). Привязки к версиям в работе, на согласовании, `REJECTED`, `CANCELLED` (и архивным
|
||
отменённым, без среза ЕХ) снимаются, статус версий не меняется.
|
||
Файл только с формами (нет документов `OSV`) удаляется целиком, как `DELETE /files/{id}`, но без запрета
|
||
`FILE_HAS_CONFIRMED_DOCUMENTS`; у смешанного файла (есть и ОСВ, и формы) удаляются только `BALANCE` / `INCOME`
|
||
(ключ `--include-mixed` — файл целиком). ОСВ вне удаляемых файлов, консолидации, справочники и аудит не трогаются.
|
||
В ядре результатов КФО ещё нет — пересчитывать после очистки нечего.
|
||
|
||
Коды выхода (P3-13): 0 — отчёт или выполнено, 1 — отказ, 2 — ошибка аргументов.
|
||
"""
|
||
|
||
import argparse
|
||
import asyncio
|
||
import sys
|
||
import uuid
|
||
from collections.abc import Callable, Sequence
|
||
from contextlib import AbstractAsyncContextManager
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
_API_DIR = Path(__file__).resolve().parents[1]
|
||
if str(_API_DIR) not in sys.path: # запуск файлом: в sys.path только `api/scripts`, а нужен `api/`
|
||
sys.path.insert(0, str(_API_DIR))
|
||
|
||
from sqlalchemy import and_, delete, exists, func, or_, select, update # noqa: E402
|
||
from sqlalchemy.ext.asyncio import AsyncSession # noqa: E402
|
||
from sqlalchemy.sql.elements import ColumnElement # noqa: E402
|
||
|
||
from src.common.clock import utc_now # noqa: E402
|
||
from src.db.db import session_transaction # noqa: E402
|
||
from src.models.audit_ex import Comment, ExSnapshot, ExSnapshotItem # noqa: E402
|
||
from src.models.consolidation import ConsolidationDocument, ConsolidationVersion # noqa: E402
|
||
from src.models.enums import ( # noqa: E402
|
||
AuditAction,
|
||
AuditSection,
|
||
ConsolidationStatus,
|
||
DocumentStatus,
|
||
DocumentType,
|
||
S3BucketRole,
|
||
)
|
||
from src.models.files import Document, StoredFile # noqa: E402
|
||
from src.repositories import files as files_repo # noqa: E402
|
||
from src.repositories import statements as statements_repo # noqa: E402
|
||
from src.services import consolidation_documents # noqa: E402
|
||
from src.services.audit import AuditService # noqa: E402
|
||
from src.services.documents import file_out # noqa: E402
|
||
from src.services.primary_files import OBJECT_TYPE, UPLOADS_PREFIX # noqa: E402
|
||
from src.services.result_mapping import RESULTS_PREFIX # noqa: E402
|
||
|
||
EXIT_OK, EXIT_REFUSED, EXIT_USAGE = 0, 1, 2
|
||
DESCRIPTION = "очистка форм перед v5.19"
|
||
FORM_TYPES = (DocumentType.BALANCE, DocumentType.INCOME)
|
||
SNAPSHOT_ITEM_STATUS = "EX_SNAPSHOT_ITEM" # «статус» блокировки, найденной по ex_snapshot_item, а не по привязке
|
||
|
||
Transaction = Callable[[], AbstractAsyncContextManager[AsyncSession]]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BlockingVersion:
|
||
"""Причина отказа: документ формы заморожен в верифицированной версии (или упомянут в срезе ЕХ)."""
|
||
|
||
version_id: uuid.UUID
|
||
status: str
|
||
document_id: uuid.UUID
|
||
|
||
|
||
@dataclass
|
||
class PurgePlan:
|
||
"""Что сделает очистка; строится `plan_purge` (только чтение)."""
|
||
|
||
form_only_files: list[uuid.UUID] = field(default_factory=list)
|
||
mixed_files: list[uuid.UUID] = field(default_factory=list)
|
||
form_documents: int = 0
|
||
confirmed_documents: int = 0
|
||
unverified_bindings: int = 0
|
||
blocking: list[BlockingVersion] = field(default_factory=list)
|
||
superseded_with_osv: list[uuid.UUID] = field(default_factory=list)
|
||
include_mixed: bool = False
|
||
|
||
|
||
class PurgeRefusedError(Exception):
|
||
"""Очистка отклонена: есть документы, замороженные в верифицированных версиях. `plan` — свежий план,
|
||
построенный под блокировками (его счётчики и `blocking` описывают состояние базы на момент отказа)."""
|
||
|
||
def __init__(self, plan: PurgePlan) -> None:
|
||
super().__init__("очистка отклонена")
|
||
self.plan = plan
|
||
self.blocking = list(plan.blocking)
|
||
|
||
|
||
def _frozen_version() -> ColumnElement[bool]:
|
||
"""Условие «версия верифицирована»: `VERIFIED`, либо `ARCHIVED` со срезом ЕХ этой версии (P3-12)."""
|
||
snapshot = exists().where(ExSnapshot.source_version_id == ConsolidationVersion.id)
|
||
return or_(
|
||
ConsolidationVersion.status == ConsolidationStatus.VERIFIED,
|
||
and_(ConsolidationVersion.status == ConsolidationStatus.ARCHIVED, snapshot),
|
||
)
|
||
|
||
|
||
async def _documents_of(
|
||
session: AsyncSession, file_ids: Sequence[uuid.UUID]
|
||
) -> list[tuple[uuid.UUID, uuid.UUID, DocumentType]]:
|
||
rows = await session.execute(
|
||
select(Document.id, Document.stored_file_id, Document.document_type).where(
|
||
Document.stored_file_id.in_(list(file_ids))
|
||
)
|
||
)
|
||
return [(row[0], row[1], row[2]) for row in rows.all()]
|
||
|
||
|
||
async def _find_blocking(session: AsyncSession, document_ids: Sequence[uuid.UUID]) -> list[BlockingVersion]:
|
||
if not document_ids:
|
||
return []
|
||
ids = list(document_ids)
|
||
bound = await session.execute(
|
||
select(ConsolidationVersion.id, ConsolidationVersion.status, ConsolidationDocument.document_id)
|
||
.join(ConsolidationDocument, ConsolidationDocument.consolidation_version_id == ConsolidationVersion.id)
|
||
.where(ConsolidationDocument.document_id.in_(ids), _frozen_version())
|
||
)
|
||
found = {BlockingVersion(version_id, status.value, document_id) for version_id, status, document_id in bound.all()}
|
||
items = await session.execute(
|
||
select(ExSnapshot.source_version_id, ExSnapshotItem.document_id)
|
||
.join(ExSnapshotItem, ExSnapshotItem.ex_snapshot_id == ExSnapshot.id)
|
||
.where(ExSnapshotItem.document_id.in_(ids))
|
||
)
|
||
found |= {BlockingVersion(version_id, SNAPSHOT_ITEM_STATUS, document_id) for version_id, document_id in items.all()}
|
||
return sorted(found, key=lambda item: (str(item.version_id), str(item.document_id), item.status))
|
||
|
||
|
||
async def plan_purge(session: AsyncSession, *, include_mixed: bool) -> PurgePlan:
|
||
"""Отбор и отчёт без изменений. «Только с формами» — нет документов `OSV`; «смешанный» — есть и ОСВ, и формы."""
|
||
forms = (
|
||
await session.execute(
|
||
select(Document.id, Document.stored_file_id, Document.status).where(Document.document_type.in_(FORM_TYPES))
|
||
)
|
||
).all()
|
||
plan = PurgePlan(include_mixed=include_mixed, form_documents=len(forms))
|
||
plan.confirmed_documents = sum(1 for _, _, status in forms if status is DocumentStatus.CONFIRMED)
|
||
file_ids = sorted({file_id for _, file_id, _ in forms})
|
||
if not file_ids:
|
||
return plan
|
||
documents = await _documents_of(session, file_ids)
|
||
with_osv = {file_id for _, file_id, kind in documents if kind is DocumentType.OSV}
|
||
plan.form_only_files = [file_id for file_id in file_ids if file_id not in with_osv]
|
||
plan.mixed_files = [file_id for file_id in file_ids if file_id in with_osv]
|
||
deleted_files = set(plan.form_only_files) | (set(plan.mixed_files) if include_mixed else set())
|
||
affected = [
|
||
document_id for document_id, file_id, kind in documents if file_id in deleted_files or (kind in FORM_TYPES)
|
||
]
|
||
plan.blocking = await _find_blocking(session, affected)
|
||
blocked_ids = {item.document_id for item in plan.blocking}
|
||
bindings = await session.execute(
|
||
select(ConsolidationDocument.document_id, func.count())
|
||
.where(ConsolidationDocument.document_id.in_(affected))
|
||
.group_by(ConsolidationDocument.document_id)
|
||
)
|
||
plan.unverified_bindings = sum(count for document_id, count in bindings.all() if document_id not in blocked_ids)
|
||
if deleted_files:
|
||
rows = await session.execute(
|
||
select(StoredFile.id).where(StoredFile.superseded_by_file_id.in_(list(deleted_files)))
|
||
)
|
||
replaced = [file_id for (file_id,) in rows.all()]
|
||
plan.superseded_with_osv = sorted(await _files_with_osv(session, replaced))
|
||
return plan
|
||
|
||
|
||
async def _files_with_osv(session: AsyncSession, file_ids: Sequence[uuid.UUID]) -> set[uuid.UUID]:
|
||
rows = await session.execute(
|
||
select(Document.stored_file_id)
|
||
.where(Document.stored_file_id.in_(list(file_ids)), Document.document_type == DocumentType.OSV)
|
||
.distinct()
|
||
)
|
||
return {file_id for (file_id,) in rows.all()}
|
||
|
||
|
||
async def _purge_whole_file(session: AsyncSession, audit: AuditService, file_id: uuid.UUID, now: datetime) -> None:
|
||
"""Файл целиком — как `PrimaryFileService.delete`, но без запрета на подтверждённые документы."""
|
||
jobs = await files_repo.lock_jobs(session, file_id)
|
||
file = await files_repo.lock_file(session, file_id)
|
||
if file is None:
|
||
return
|
||
documents = await files_repo.lock_file_documents(session, file_id)
|
||
bound = await consolidation_documents.bound_versions(session, documents=documents)
|
||
await consolidation_documents.unbind_documents(session, document_ids=[document.id for document in documents])
|
||
await session.execute(
|
||
update(StoredFile).where(StoredFile.superseded_by_file_id == file_id).values(superseded_by_file_id=None)
|
||
)
|
||
current = next((job for job in jobs if job.id == file.current_recognition_job_id), None)
|
||
counts = await files_repo.comments_count_by_document(session, [document.id for document in documents])
|
||
before = file_out(file, current, documents, counts).model_dump(mode="json")
|
||
object_ids = [file.s3_object_id, *(job.result_s3_object_id for job in jobs if job.result_s3_object_id)]
|
||
await files_repo.delete_file_rows(session, file)
|
||
owned_prefixes = [
|
||
(S3BucketRole.uploads, f"{UPLOADS_PREFIX}/{file_id}/"),
|
||
(S3BucketRole.results, f"{RESULTS_PREFIX}/{file_id}/"),
|
||
]
|
||
await files_repo.expire_s3_objects(session, object_ids, owned_prefixes=owned_prefixes, now=now)
|
||
await audit.record(
|
||
action=AuditAction.FILE_DELETE,
|
||
section=AuditSection.PRIMARY_DOCUMENTS,
|
||
object_type=OBJECT_TYPE,
|
||
object_id=file_id,
|
||
user_id=None,
|
||
description=DESCRIPTION,
|
||
before_value=before,
|
||
after_value={"unbound_version_ids": sorted({version_id for version_id, _ in bound})},
|
||
)
|
||
|
||
|
||
async def _purge_forms_of_file(session: AsyncSession, audit: AuditService, file_id: uuid.UUID) -> None:
|
||
"""Смешанный файл: удаляются только `BALANCE` / `INCOME`; файл, ОСВ и задания остаются."""
|
||
jobs = await files_repo.lock_jobs(session, file_id)
|
||
file = await files_repo.lock_file(session, file_id)
|
||
if file is None:
|
||
return
|
||
documents = await files_repo.lock_file_documents(session, file_id)
|
||
forms = [document for document in documents if document.document_type in FORM_TYPES]
|
||
if not forms:
|
||
return
|
||
form_ids = [document.id for document in forms]
|
||
bound = await consolidation_documents.bound_versions(session, documents=forms)
|
||
await consolidation_documents.unbind_documents(session, document_ids=form_ids)
|
||
current = next((job for job in jobs if job.id == file.current_recognition_job_id), None)
|
||
counts = await files_repo.comments_count_by_document(session, [document.id for document in documents])
|
||
before = file_out(file, current, documents, counts).model_dump(mode="json")
|
||
await statements_repo.delete_documents_statements(session, form_ids)
|
||
await session.execute(
|
||
delete(Comment).where(Comment.object_type == files_repo.COMMENT_OBJECT_TYPE, Comment.object_id.in_(form_ids))
|
||
)
|
||
await session.execute(delete(Document).where(Document.id.in_(form_ids)))
|
||
await audit.record(
|
||
action=AuditAction.FILE_DELETE,
|
||
section=AuditSection.PRIMARY_DOCUMENTS,
|
||
object_type=OBJECT_TYPE,
|
||
object_id=file_id,
|
||
user_id=None,
|
||
description=DESCRIPTION,
|
||
before_value=before,
|
||
after_value={
|
||
"deleted_documents": [str(document_id) for document_id in form_ids],
|
||
"unbound_version_ids": sorted({version_id for version_id, _ in bound}),
|
||
},
|
||
)
|
||
|
||
|
||
async def apply_purge(session: AsyncSession, plan: PurgePlan, *, now: datetime) -> None:
|
||
"""Выполняет очистку в транзакции сессии (коммит — за вызывающим). Порядок блокировок: файл → документы → версии.
|
||
|
||
Блокирует файлы плана (по `id`), заново проверяет отказ под блокировками и только потом меняет данные: при
|
||
`PurgeRefusedError` база не изменена."""
|
||
locked: set[uuid.UUID] = set()
|
||
|
||
async def lock(file_ids: Sequence[uuid.UUID]) -> None:
|
||
for file_id in sorted(set(file_ids) - locked):
|
||
await files_repo.lock_jobs(session, file_id)
|
||
await files_repo.lock_file(session, file_id)
|
||
await files_repo.lock_file_documents(session, file_id)
|
||
locked.add(file_id)
|
||
|
||
await lock([*plan.form_only_files, *plan.mixed_files])
|
||
fresh = await plan_purge(session, include_mixed=plan.include_mixed) # T4.2: списки файлов — из свежего плана
|
||
await lock([*fresh.form_only_files, *fresh.mixed_files]) # файлы, появившиеся после первого плана
|
||
if fresh.blocking:
|
||
raise PurgeRefusedError(fresh)
|
||
whole = list(fresh.form_only_files) + (list(fresh.mixed_files) if fresh.include_mixed else [])
|
||
partial = [] if fresh.include_mixed else list(fresh.mixed_files)
|
||
audit = AuditService(session)
|
||
for file_id in sorted(whole):
|
||
await _purge_whole_file(session, audit, file_id, now)
|
||
for file_id in sorted(partial):
|
||
await _purge_forms_of_file(session, audit, file_id)
|
||
|
||
|
||
def render_report(plan: PurgePlan) -> str:
|
||
lines = [
|
||
f"файлов только с формами: {len(plan.form_only_files)}",
|
||
f"смешанных файлов: {len(plan.mixed_files)}"
|
||
+ (" (будут удалены целиком, --include-mixed)" if plan.include_mixed else " (удаляются только формы)"),
|
||
f"документов форм: {plan.form_documents}",
|
||
f"подтверждённых документов: {plan.confirmed_documents}",
|
||
f"привязок к неверифицированным версиям: {plan.unverified_bindings}",
|
||
]
|
||
if plan.mixed_files:
|
||
lines.append("смешанные файлы (повторная загрузка невозможна — дедуп по sha256):")
|
||
lines.extend(f" {file_id}" for file_id in plan.mixed_files)
|
||
if plan.superseded_with_osv:
|
||
lines.append("заменённые удаляемыми файлами файлы с ОСВ (остаются SUPERSEDED):")
|
||
lines.extend(f" {file_id}" for file_id in plan.superseded_with_osv)
|
||
if plan.blocking:
|
||
lines.append("ОТКАЗ: формы заморожены в верифицированных версиях, решение — за владельцем:")
|
||
lines.extend(
|
||
f" версия {item.version_id} ({item.status}), документ {item.document_id}" for item in plan.blocking
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _parse(argv: Sequence[str]) -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(prog="purge_statements", description="Очистка форм ББ/ОФР перед выкладкой v5.19")
|
||
mode = parser.add_mutually_exclusive_group()
|
||
mode.add_argument("--dry-run", action="store_true", help="только отчёт (по умолчанию)")
|
||
mode.add_argument("--apply", action="store_true", help="выполнить очистку одной транзакцией")
|
||
parser.add_argument("--include-mixed", action="store_true", help="смешанные файлы удалить целиком")
|
||
return parser.parse_args(list(argv))
|
||
|
||
|
||
async def run(argv: Sequence[str], *, transaction: Transaction = session_transaction) -> int:
|
||
"""Точка входа; `transaction` — фабрика транзакции (тесты подставляют свою сессию)."""
|
||
try:
|
||
args = _parse(argv)
|
||
except SystemExit as error:
|
||
return EXIT_USAGE if error.code else EXIT_OK
|
||
try:
|
||
async with transaction() as session:
|
||
plan = await plan_purge(session, include_mixed=args.include_mixed)
|
||
print(render_report(plan))
|
||
if not args.apply:
|
||
return EXIT_OK
|
||
if plan.blocking:
|
||
print("очистка отклонена, база не изменена", file=sys.stderr)
|
||
return EXIT_REFUSED
|
||
await apply_purge(session, plan, now=utc_now())
|
||
except PurgeRefusedError as error:
|
||
print(render_report(error.plan), file=sys.stderr) # T4.4: свежий план с реальными счётчиками
|
||
print("очистка отклонена, база не изменена", file=sys.stderr)
|
||
return EXIT_REFUSED
|
||
print("очистка выполнена")
|
||
return EXIT_OK
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(asyncio.run(run(sys.argv[1:])))
|