"""Фаза 2: применение проверенного набора к БД — upsert по натуральному ключу, без удалений и без commit.""" import datetime import uuid from typing import Any from sqlalchemy import func, select, update from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.ext.asyncio import AsyncSession from scripts.seed_lib.apply_rules import apply_elimination_rules, apply_required_document_rules from scripts.seed_lib.model import SeedData, SeedError, SeedReport, TableStat from src.domain.aliases import normalize_alias from src.models.access import AppUser, Permission, Role, RolePermission, UserRole from src.models.accounting import ChartOfAccount from src.models.directories import ( COMPANY_GROUP_CODE_SEQ, RELATED_GROUP_CODE_SEQ, CompanyGroup, LegalEntity, LegalEntityAlias, RegionalOffice, RelatedGroup, RelatedGroupMember, ) from src.repositories.directories import next_code from src.services.company_groups import COMPANY_GROUP_CODE_PREFIX from src.services.related_groups import CODE_PREFIXES async def _upsert( session: AsyncSession, model: Any, key: str, rows: list[dict[str, Any]], stat: TableStat, ) -> set[Any]: """Вставить новые и обновить изменившиеся строки по натуральному ключу `key`; удалений нет. Возвращает ключи строк, ВСТАВЛЕННЫХ этим прогоном (для матрицы прав: связи новых ролей и прав). """ if not rows: return set() key_column = getattr(model, key) value_names = [name for name in rows[0] if name != key] existing: dict[Any, tuple[Any, ...]] = { row[0]: tuple(row[1:]) for row in (await session.execute(select(key_column, *(getattr(model, name) for name in value_names)))).all() } changed = [] inserted: set[Any] = set() for row in rows: current = existing.get(row[key]) if current is None: stat.inserted += 1 inserted.add(row[key]) changed.append(row) elif current != tuple(row[name] for name in value_names): stat.updated += 1 changed.append(row) else: stat.unchanged += 1 if changed: statement = pg_insert(model) statement = statement.on_conflict_do_update( index_elements=[key_column], set_={name: statement.excluded[name] for name in value_names} ) await session.execute(statement, changed) return inserted MATRIX_SKIPPED = ( "матрица прав уже заполнена; seed не возвращает права, снятые администратором, и добавляет только связи ролей " "и прав, впервые созданных этим запуском (правка — PUT /admin/roles/{code}/permissions)" ) async def _apply_role_permissions( # noqa: PLR0913 - набор, статистика и коды впервые вставленных ролей и прав session: AsyncSession, data: SeedData, stat: TableStat, *, new_roles: set[str], new_permissions: set[str], ) -> None: """Матрица «роль × право» из файла (issue #9 п.7, ревью R-3). Пустая `role_permission` (первичная загрузка, в том числе после того, как администратор снял все связи) — вставляется вся матрица файла. Непустая — матрицей владеет администратор: вставляются только пары, у которых роль ИЛИ право вставлены этим же прогоном (`new_roles`/`new_permissions` — их администратор снять не мог); остальные пары не трогаются (`stat.skipped_reason`): снятые права не возвращаются, добавленные не отзываются. """ matrix_filled = await session.scalar(select(RolePermission.role_id).limit(1)) is not None if matrix_filled: stat.skipped_reason = MATRIX_SKIPPED role_ids = {code: id_ for id_, code in (await session.execute(select(Role.id, Role.code))).all()} permission_ids = {code: id_ for id_, code in (await session.execute(select(Permission.id, Permission.code))).all()} new_rows = [ { "role_id": role_ids[role_code], "permission_id": permission_ids[code], "granted_by": None, "granted_at": func.now(), } for role_code, codes in data.role_permissions.items() for code in codes if not matrix_filled or role_code in new_roles or code in new_permissions ] if new_rows: stat.inserted += len(new_rows) # .values(): func.now() допустим только во встроенном VALUES, не в executemany statement = ( pg_insert(RolePermission) .values(new_rows) .on_conflict_do_nothing(index_elements=["role_id", "permission_id"]) ) await session.execute(statement) async def _apply_chart_of_accounts(session: AsyncSession, data: SeedData, stat: TableStat) -> None: """План счетов: upsert по `code` в порядке глубины (родитель раньше потомков), одним `run_seed` и одной транзакцией. Идентификаторы новых строк генерирует приложение заранее, поэтому `parent_id` известен при вставке и второй проход не нужен; существующим строкам при расхождении с файлом обновляются поля и `parent_id`. Удалений нет. """ if not data.accounts: return stored = ( await session.execute( select( ChartOfAccount.code, ChartOfAccount.id, ChartOfAccount.parent_id, ChartOfAccount.name, ChartOfAccount.kind, ChartOfAccount.row_type, ChartOfAccount.is_off_balance, ) ) ).all() ids = {row.code: row.id for row in stored} existing = {row.code: (row.parent_id, row.name, row.kind, row.row_type, row.is_off_balance) for row in stored} for code in data.accounts: ids.setdefault(code, uuid.uuid4()) rows = [] for code, account in sorted(data.accounts.items(), key=lambda item: item[0].count(".")): parent_id = ids[account.parent_code] if account.parent_code is not None else None values = (parent_id, account.name, account.kind, account.row_type, account.is_off_balance) if code not in existing: stat.inserted += 1 elif existing[code] != values: stat.updated += 1 else: stat.unchanged += 1 continue rows.append( { "id": ids[code], "code": code, "parent_id": parent_id, "name": account.name, "kind": account.kind, "row_type": account.row_type, "is_off_balance": account.is_off_balance, } ) if rows: statement = pg_insert(ChartOfAccount) statement = statement.on_conflict_do_update( index_elements=[ChartOfAccount.code], set_={ name: statement.excluded[name] for name in ("parent_id", "name", "kind", "row_type", "is_off_balance") }, ) await session.execute(statement, rows) async def _apply_users(session: AsyncSession, data: SeedData, user_stat: TableStat, link_stat: TableStat) -> None: """Dev-пользователи: upsert `app_user` по `email`, затем `user_role` по (user_id, role_id). Email из файла нормализуется (`strip().lower()`, DBML v5.15) — и при сверке с БД, и при вставке. `created_at` ставится при вставке, `updated_at` — при вставке и при изменении данных (триггера в БД нет); неизменившиеся строки не трогаются. Назначения ролей только добавляются: роль, убранная из файла, остаётся назначенной (удалений нет). `external_id` не задаётся (SSO-субъект dev-пользователя неизвестен). """ if not data.users: return office_ids = { name: id_ for id_, name in (await session.execute(select(RegionalOffice.id, RegionalOffice.name))).all() } role_ids = {code: id_ for id_, code in (await session.execute(select(Role.id, Role.code))).all()} stored = { (row.email or "").lower(): row for row in ( await session.execute( select( AppUser.id, AppUser.email, AppUser.full_name, AppUser.employee_no, AppUser.regional_office_id, AppUser.business_segment, AppUser.is_active, ) ) ).all() } user_ids: dict[str, Any] = {} users = [(user.email.strip().lower(), user) for user in data.users] # email нормализуется один раз for email, user in users: values = { "full_name": user.full_name, "employee_no": user.employee_no, "regional_office_id": office_ids[user.regional_office], "business_segment": user.business_segment, "is_active": user.is_active, } row = stored.get(email) if row is None: user_stat.inserted += 1 user_ids[email] = uuid.uuid4() await session.execute( pg_insert(AppUser).values( id=user_ids[email], email=email, **values, created_at=func.now(), updated_at=func.now() ) ) continue user_ids[email] = row.id if any(getattr(row, name) != value for name, value in values.items()): user_stat.updated += 1 await session.execute(update(AppUser).where(AppUser.id == row.id).values(**values, updated_at=func.now())) else: user_stat.unchanged += 1 existing = set((await session.execute(select(UserRole.user_id, UserRole.role_id))).all()) new_rows = [] for email, user in users: for role_code in user.role_codes: pair = (user_ids[email], role_ids[role_code]) if pair in existing: link_stat.unchanged += 1 else: link_stat.inserted += 1 new_rows.append({"id": uuid.uuid4(), "user_id": pair[0], "role_id": pair[1]}) if new_rows: await session.execute( pg_insert(UserRole).on_conflict_do_nothing(index_elements=["user_id", "role_id"]), new_rows ) async def _apply_company_groups(session: AsyncSession, data: SeedData, report: SeedReport) -> None: # noqa: PLR0912, PLR0915 - пять таблиц одним проходом по набору """Dev-ГК: upsert `company_group` по `name`, `legal_entity` по `inn`, `related_group` по (ГК, name); алиасы и члены групп только добавляются. Коды — из тех же последовательностей, что у API (`next_code`); истории и аудита нет, как у импорта ЕЦП при создании; `updated_by`/`created_by` пусты (seed — не пользователь). Удалений нет; вид существующей ГСК/ГСЗ seed не меняет. """ if not data.company_groups: return now = datetime.datetime.now(datetime.UTC) stats = { name: report.stat(name) for name in ("company_group", "legal_entity", "legal_entity_alias", "related_group", "related_group_member") } offices = {name: id_ for id_, name in (await session.execute(select(RegionalOffice.id, RegionalOffice.name))).all()} groups = { row.name: row for row in ( await session.execute( select(CompanyGroup.id, CompanyGroup.name, CompanyGroup.crm_id, CompanyGroup.base_regional_office_id) ) ).all() } entities = { row.inn: row for row in ( await session.execute( select( LegalEntity.id, LegalEntity.inn, LegalEntity.name, LegalEntity.entity_type, LegalEntity.company_group_id, LegalEntity.regional_office_id, ) ) ).all() } alias_owner = { normalized: entity_id for entity_id, normalized in ( await session.execute(select(LegalEntityAlias.legal_entity_id, LegalEntityAlias.normalized_alias)) ).all() } related = { (row.company_group_id, row.name): row for row in ( await session.execute( select(RelatedGroup.id, RelatedGroup.company_group_id, RelatedGroup.name, RelatedGroup.kind) ) ).all() } members: set[tuple[uuid.UUID, uuid.UUID]] = { (row.related_group_id, row.legal_entity_id) for row in ( await session.execute(select(RelatedGroupMember.related_group_id, RelatedGroupMember.legal_entity_id)) ).all() } for group in data.company_groups: office_id = offices[group.regional_office] stored_group = groups.get(group.name) if stored_group is None: group_id = uuid.uuid4() session.add( CompanyGroup( id=group_id, code=await next_code(session, COMPANY_GROUP_CODE_SEQ, COMPANY_GROUP_CODE_PREFIX), crm_id=group.crm_id, name=group.name, base_regional_office_id=office_id, valid_from=now, created_at=now, updated_at=now, ) ) stats["company_group"].inserted += 1 else: group_id = stored_group.id if (stored_group.crm_id, stored_group.base_regional_office_id) != (group.crm_id, office_id): await session.execute( update(CompanyGroup) .where(CompanyGroup.id == group_id) .values(crm_id=group.crm_id, base_regional_office_id=office_id, updated_at=now) ) stats["company_group"].updated += 1 else: stats["company_group"].unchanged += 1 await session.flush() entity_ids: dict[str, uuid.UUID] = {} for entity in group.legal_entities: values: dict[str, Any] = { "name": entity.name, "entity_type": entity.entity_type, "company_group_id": group_id, "regional_office_id": office_id, } stored_entity = entities.get(entity.inn) if stored_entity is None: entity_id = uuid.uuid4() session.add( LegalEntity(id=entity_id, inn=entity.inn, **values, valid_from=now, created_at=now, updated_at=now) ) stats["legal_entity"].inserted += 1 else: entity_id = stored_entity.id if any(getattr(stored_entity, name) != value for name, value in values.items()): await session.execute( update(LegalEntity).where(LegalEntity.id == entity_id).values(**values, updated_at=now) ) stats["legal_entity"].updated += 1 else: stats["legal_entity"].unchanged += 1 entity_ids[entity.inn] = entity_id for alias in entity.aliases: normalized = normalize_alias(alias) owner = alias_owner.get(normalized) if owner == entity_id: stats["legal_entity_alias"].unchanged += 1 continue if owner is not None: raise SeedError( f"алиас «{alias}» ЮЛ {entity.inn} уже закреплён за другим ЮЛ (id {owner}); перенос — через API" ) session.add( LegalEntityAlias( legal_entity_id=entity_id, alias=alias, normalized_alias=normalized, created_at=now ) ) alias_owner[normalized] = entity_id stats["legal_entity_alias"].inserted += 1 await session.flush() for related_group in group.related_groups: stored_related = related.get((group_id, related_group.name)) if stored_related is None: related_id = uuid.uuid4() session.add( RelatedGroup( id=related_id, code=await next_code(session, RELATED_GROUP_CODE_SEQ, CODE_PREFIXES[related_group.kind]), company_group_id=group_id, kind=related_group.kind, name=related_group.name, created_date=now.date(), ) ) stats["related_group"].inserted += 1 else: related_id = stored_related.id stats["related_group"].unchanged += 1 await session.flush() for inn in related_group.member_inns: pair = (related_id, entity_ids[inn]) if pair in members: stats["related_group_member"].unchanged += 1 continue session.add( RelatedGroupMember( related_group_id=related_id, legal_entity_id=entity_ids[inn], kind=related_group.kind, valid_from=now, valid_to=None, is_current=True, ) ) members.add(pair) stats["related_group_member"].inserted += 1 await session.flush() async def apply_seed(session: AsyncSession, data: SeedData) -> SeedReport: """Фаза 2: upsert набора в порядке зависимостей. Без commit — транзакцией управляет вызывающий.""" report = SeedReport() await _upsert( session, RegionalOffice, "name", [{"name": o.name, "is_head_office": o.is_head_office, "is_active": o.is_active} for o in data.regional_offices], report.stat("regional_office"), ) new_roles = await _upsert( session, Role, "code", [ { "code": r.code, "name": r.name, "description": r.description, "is_system": r.is_system, "is_active": r.is_active, } for r in data.roles ], report.stat("role"), ) new_permissions = await _upsert( session, Permission, "code", [ { "code": p.code, "object_type": p.object_type, "action": p.action, "name": p.name, "description": p.description, } for p in data.permissions ], report.stat("permission"), ) await _apply_role_permissions( session, data, report.stat("role_permission"), new_roles=new_roles, new_permissions=new_permissions ) await _apply_chart_of_accounts(session, data, report.stat("chart_of_account")) await apply_elimination_rules( session, data, report.stat("elimination_rule"), report.stat("elimination_rule_account") ) await apply_required_document_rules(session, data, report.stat("required_document_rule")) # dev-пользователи — после ролей и филиалов (в том числе dev-филиалов, они входят в `regional_offices`) await _apply_users(session, data, report.stat("app_user"), report.stat("user_role")) # dev-ГК — после филиалов (ссылки по имени); справочники независимы от пользователей await _apply_company_groups(session, data, report) return report