scripts/fiscalization_flow.py - master - beyondviolet/edw_prefect

scripts/fiscalization_flow.py
359 строк12 KB
get458
first_commit
04 авг 2026, 07:38
04 авг 2026, 07:38b3e3bb3
import asyncio
import logging
import traceback
from uuid import UUID

from bvcore.app import set_cwd
from bvcore.db import DBPool
from bvcore.db.models import Model
from prefect import flow, task

set_cwd()

from lib.db import init_db_pool_or_raise
from lib.fiscalization_model import FiscalizationModel
from lib.fiscalization_service import FiscalizationService
from lib.notify import tg_notify
from lib.setup import setup

bv_log = logging.getLogger("bviolet-fiscal")
bv_log.setLevel(logging.INFO)

handler = logging.StreamHandler()
handler.setFormatter(
    logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
)

if not bv_log.handlers:
    bv_log.addHandler(handler)

db = DBPool("config/database/production.yml")


@task(name="Init DB and Config", retries=3, retry_delay_seconds=5)
async def init_app(sdk_config_path: str = "config/bv_sdk-prod.yml"):
    await init_db_pool_or_raise(db, logger=bv_log)
    await setup(sdk_config_path)
    bv_log.info("DB and payment config initialized")


@task(name="Fetch Payments Needing Closing Receipt", retries=2, retry_delay_seconds=10)
async def fetch_payments_needing_closing_receipt(limit: int = 100) -> list[dict]:
    pool = db("default")
    sql = """
        SELECT p.uuid AS payment_uuid, p.status, pg.code AS gateway_code
        FROM payments.payments p
        JOIN payments.gateways pg ON pg.id = p.gateway_id
        WHERE p.status = 'succeeded'
          AND pg.gw_type != 'tochka_qr'
          AND NOT EXISTS (
              SELECT 1
              FROM payments.receipts r
              WHERE r.payment_uuid = p.uuid
                AND r.receipt_type = 'sale'
                AND r.receipt_variant = 'closing'
          )
          AND p.updated_at >= NOW() - INTERVAL '7 days'
        ORDER BY p.updated_at DESC
        LIMIT $1
    """
    async with pool.acquire() as conn:
        rows = await conn.fetch(sql, limit)

    result = [dict(r) for r in rows]
    bv_log.info(f"Payments needing closing receipt: {len(result)}")
    return result


@task(name="Create Closing Receipt", retries=2, retry_delay_seconds=10)
async def create_closing_receipt(payment_uuid: UUID) -> bool:
    try:
        await FiscalizationModel().create_closing_receipt(payment_uuid)
        bv_log.info(f"[{payment_uuid}] Closing receipt created")
        return True
    except Exception:
        bv_log.error(f"[{payment_uuid}] Failed to create closing receipt")
        bv_log.error(traceback.format_exc())
        return False


@task(name="Fetch Refunds Needing Receipt", retries=2, retry_delay_seconds=10)
async def fetch_refunds_needing_receipt(limit: int = 100) -> list[dict]:
    pool = db("default")
    sql = """
        SELECT pr.uuid AS refund_uuid, pr.payment_uuid
        FROM payments.refunds pr
        JOIN payments.payments pp ON pp.uuid = pr.payment_uuid
        JOIN payments.gateways pg ON pg.id = pp.gateway_id
        WHERE pr.status = 'succeeded'
          AND pg.gw_type != 'tochka_qr'
          AND NOT EXISTS (
              SELECT 1
              FROM payments.receipts r
              WHERE r.refund_uuid = pr.uuid
                AND r.receipt_type = 'refund'
                AND r.receipt_variant = 'refund'
          )
          AND pr.updated_at >= NOW() - INTERVAL '7 days'
        ORDER BY pr.updated_at DESC
        LIMIT $1
    """
    async with pool.acquire() as conn:
        rows = await conn.fetch(sql, limit)

    result = [dict(r) for r in rows]
    bv_log.info(f"Refunds needing receipt: {len(result)}")
    return result


@task(name="Create Refund Receipt", retries=2, retry_delay_seconds=10)
async def create_refund_receipt(refund_uuid: UUID) -> bool:
    try:
        await FiscalizationModel().create_refund_receipt(refund_uuid)
        bv_log.info(f"[{refund_uuid}] Refund receipt created")
        return True
    except Exception:
        bv_log.error(f"[{refund_uuid}] Failed to create refund receipt")
        bv_log.error(traceback.format_exc())
        return False


@task(name="Fetch Pending Receipts", retries=2, retry_delay_seconds=10)
async def fetch_pending_receipts(limit: int = 45) -> list[dict]:
    pool = db("default")

    async with pool.acquire() as conn:
        async with conn.transaction():
            rows = await conn.fetch(
                """
                WITH locked AS (
                    SELECT pr.uuid, pr.external_code
                    FROM payments.receipts pr
                    JOIN payments.payments p ON p.uuid = pr.payment_uuid
                    JOIN payments.gateways pg ON p.gateway_id = pg.id
                    WHERE pr.status = 'pending'
                      AND pr.error_code IS NULL
                      AND pr.is_processing = FALSE
                      AND (
                          pr.receipt_type = 'sale'
                          OR (pr.receipt_type = 'refund' AND p.status IN ('succeeded', 'canceled'))
                      )
                      AND pr.receipt_variant IN ('closing', 'refund', 'full_refund')
                      AND pg.gw_type != 'tochka_qr'
                    ORDER BY pr.created_at
                    LIMIT $1
                    FOR UPDATE SKIP LOCKED
                )
                UPDATE payments.receipts
                SET is_processing = TRUE,
                    processing_started_at = NOW()
                FROM locked
                WHERE payments.receipts.uuid = locked.uuid
                RETURNING payments.receipts.uuid, locked.external_code
                """,
                limit,
            )

    result = [dict(r) for r in rows]
    bv_log.info(f"Pending receipts to process: {len(result)}")
    return result


@task(name="Send or Poll Receipt", retries=1, retry_delay_seconds=15)
async def send_or_poll_receipt(receipt_uuid: UUID, external_code) -> bool:
    service = FiscalizationService()

    try:
        if not external_code:
            sent = await service.send(receipt_uuid)
            if not sent:
                bv_log.error(f"[{receipt_uuid}] Failed to send to Atol")
                return False
            bv_log.info(f"[{receipt_uuid}] Sent to Atol")
        else:
            ok = await service.get_status(receipt_uuid)
            if not ok:
                bv_log.error(f"[{receipt_uuid}] Failed to poll status (external={external_code})")
                return False
            bv_log.info(f"[{receipt_uuid}] Status polled")

        return True

    except Exception:
        bv_log.error(f"[{receipt_uuid}] Error during send/poll")
        bv_log.error(traceback.format_exc())
        return False

    finally:
        pool = db("default")
        await pool.execute(
            """
            UPDATE payments.receipts
            SET is_processing = FALSE,
                processing_started_at = NULL
            WHERE uuid = $1
            """,
            receipt_uuid,
        )


@task(name="Release Stale Locks")
async def release_stale_locks(timeout_minutes: int = 15):
    pool = db("default")

    rows = await pool.fetch(
        """
        SELECT uuid, external_code
        FROM payments.receipts
        WHERE is_processing = TRUE
          AND processing_started_at < NOW() - ($1 * INTERVAL '1 minute')
        """,
        timeout_minutes,
    )

    if not rows:
        return 0

    service = FiscalizationService()

    for row in rows:
        receipt_uuid = row["uuid"]
        external_code = row["external_code"]

        if external_code:
            try:
                ok = await service.get_status(receipt_uuid)
                if ok:
                    bv_log.info(f"[{receipt_uuid}] Stale lock: status polled from Atol")
                else:
                    bv_log.warning(f"[{receipt_uuid}] Stale lock: Atol poll failed, releasing lock")
            except Exception:
                bv_log.error(f"[{receipt_uuid}] Stale lock: error polling Atol")
                bv_log.error(traceback.format_exc())
        else:
            bv_log.warning(f"[{receipt_uuid}] Stale lock: no external_code, releasing lock")

    released = await pool.fetchval(
        """
        WITH updated AS (
            UPDATE payments.receipts
            SET is_processing = FALSE,
                processing_started_at = NULL
            WHERE is_processing = TRUE
              AND processing_started_at < NOW() - ($1 * INTERVAL '1 minute')
            RETURNING uuid
        )
        SELECT COUNT(*) FROM updated
        """,
        timeout_minutes,
    )

    if released:
        bv_log.warning(f"Released {released} stale processing locks")

    return released or 0


async def send_flow_summary(
    stale_released: int,
    closing_payments: list[dict],
    closing_created: int,
    closing_errors: int,
    refunds: list[dict],
    refund_created: int,
    refund_errors: int,
    pending: list[dict],
    processed: int,
    poll_errors: int,
):
    total_errors = closing_errors + refund_errors + poll_errors

    status_text = "успешно завершён" if total_errors == 0 else "завершён с ошибками"

    message = (
        f"Флоу фискализации {status_text}\n"
        f"Снято зависших блокировок: {stale_released}\n"
        f"Чеки закрытия: найдено {len(closing_payments)}, создано {closing_created}, ошибок {closing_errors}\n"
        f"Чеки возврата: найдено {len(refunds)}, создано {refund_created}, ошибок {refund_errors}\n"
        f"Отправка/опрос чеков: найдено {len(pending)}, обработано {processed}, ошибок {poll_errors}"
    )

    await tg_notify(message)


@flow(name="Fiscalization Flow")
async def fiscalization_flow(
    limit: int = 100,
    sdk_config_path: str = "config/bv_sdk-prod.yml",
):
    try:
        await init_app(sdk_config_path)

        stale_released = await release_stale_locks(timeout_minutes=10)

        closing_payments = await fetch_payments_needing_closing_receipt(limit=limit)
        closing_created = 0
        closing_errors = 0

        for payment in closing_payments:
            ok = await create_closing_receipt(payment["payment_uuid"])
            if ok:
                closing_created += 1
            else:
                closing_errors += 1

        refunds = await fetch_refunds_needing_receipt(limit=limit)
        refund_created = 0
        refund_errors = 0

        for refund in refunds:
            ok = await create_refund_receipt(refund["refund_uuid"])
            if ok:
                refund_created += 1
            else:
                refund_errors += 1

        pending = await fetch_pending_receipts(limit=45)
        processed = 0
        poll_errors = 0

        for receipt in pending:
            ok = await send_or_poll_receipt(receipt["uuid"], receipt["external_code"])
            if ok:
                processed += 1
            else:
                poll_errors += 1

        total_errors = closing_errors + refund_errors + poll_errors

        if total_errors > 0:
            await tg_notify(
                "Флоу фискализации завершён с ошибками\n"
                f"Снято зависших блокировок: {stale_released}\n"
                f"Чеки закрытия: найдено {len(closing_payments)}, создано {closing_created}, ошибок {closing_errors}\n"
                f"Чеки возврата: найдено {len(refunds)}, создано {refund_created}, ошибок {refund_errors}\n"
                f"Отправка/опрос чеков: найдено {len(pending)}, обработано {processed}, ошибок {poll_errors}"
            )

        # await send_flow_summary(
        #     stale_released=stale_released,
        #     closing_payments=closing_payments,
        #     closing_created=closing_created,
        #     closing_errors=closing_errors,
        #     refunds=refunds,
        #     refund_created=refund_created,
        #     refund_errors=refund_errors,
        #     pending=pending,
        #     processed=processed,
        #     poll_errors=poll_errors,
        # )

    except Exception:
        err = traceback.format_exc()
        bv_log.error(f"Fiscalization Flow failed:\n{err}")
        await tg_notify(f"❌ Флоу фискализации завершился с критической ошибкой:\n{err[:3000]}")
        raise


if __name__ == "__main__":
    asyncio.run(fiscalization_flow())