scripts/refiscalize_flow.py - master - beyondviolet/edw_prefect

scripts/refiscalize_flow.py
153 строки4 KB
get458
first_commit
04 авг 2026, 07:38
04 авг 2026, 07:38b3e3bb3
import logging
import traceback

import pandas as pd
from prefect import flow, task

from bvcore.db import DBPool
from bvcore.app import set_cwd

from lib.db import init_db_pool_or_raise
from lib.fiscalization_service import FiscalizationService
from lib.setup import setup

set_cwd()

bv_log = logging.getLogger("bviolet")
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")
RESETTABLE_ERROR_CODES = ("1", "50", "421")


@task
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
async def fetch_receipts(limit: int = 10):
    await db.init_pool()
    pool = db("default")

    sql_query = """
        SELECT pr.uuid, pr.external_code, pr.error_code
        from payments.payments pp
        join payments.receipts pr on pr.payment_uuid=pp.uuid
        where pp.status='succeeded' and pr.status='canceled' and pr.receipt_variant='closing' and pr.resend_count<10
        and pp.created_at>=current_date-7
        LIMIT $1;
    """

    async with pool.acquire() as conn:
        rows = await conn.fetch(sql_query, limit)

    receipts = [dict(r) for r in rows]
    bv_log.info(f"Fetched {len(receipts)} receipts to resend.")
    return receipts


@task
async def reset_receipt(rcpt_uuid: str):
    await db.init_pool()
    pool = db("default")

    sql = """
        UPDATE payments.receipts
        SET uuid = default,
            external_code = NULL,
            status = 'pending',
            fiscal_provider_id = NULL,
            fiscal_receipt_number = NULL,
            fiscal_storage_number = NULL,
            fiscal_document_number = NULL,
            fiscal_document_attribute = NULL,
            error_code = NULL,
            resend_count = resend_count + 1
        WHERE uuid = $1
          AND error_code = ANY($2::text[])
    """

    async with pool.acquire() as conn:
        status = await conn.execute(sql, rcpt_uuid, list(RESETTABLE_ERROR_CODES))

    updated = status.endswith(" 1")
    if updated:
        bv_log.info(f"Receipt {rcpt_uuid} reset to pending.")
    else:
        bv_log.warning(
            f"Receipt {rcpt_uuid} not updated (maybe already processed or conditions changed)."
        )
    return updated


@task
async def refresh_receipt_error_from_atol(rcpt_uuid: str, external_code: str) -> bool:
    if not external_code:
        return False

    service = FiscalizationService()

    try:
        ok = await service.get_status(rcpt_uuid)
        if ok:
            bv_log.info(
                f"Receipt {rcpt_uuid}: ATOL status checked by external_code={external_code}."
            )
        else:
            bv_log.warning(
                f"Receipt {rcpt_uuid}: ATOL status check returned no actionable result "
                f"(external_code={external_code})."
            )
        return ok
    except Exception:
        bv_log.error(
            f"Receipt {rcpt_uuid}: error while checking ATOL status by "
            f"external_code={external_code}."
        )
        bv_log.error(traceback.format_exc())
        return False


@flow(name="Resend Canceled Receipts")
async def resend_receipts_flow(
    limit: int = 10,
    sdk_config_path: str = "config/bv_sdk-prod.yml",
):
    await init_app(sdk_config_path)
    receipts = await fetch_receipts(limit=limit)

    processed_results = []
    for rcpt in receipts:
        receipt_uuid = rcpt["uuid"]
        error_code = rcpt.get("error_code")
        external_code = rcpt.get("external_code")

        if error_code not in RESETTABLE_ERROR_CODES and external_code:
            await refresh_receipt_error_from_atol(receipt_uuid, external_code)

        ok = await reset_receipt(receipt_uuid)
        processed_results.append({"uuid": receipt_uuid, "updated": ok})

    df = pd.DataFrame(processed_results)
    updated_count = int(df["updated"].sum()) if not df.empty else 0

    bv_log.info(f"Flow finished. Updated: {updated_count} / {len(df)}")
    print(df)

    return df


if __name__ == "__main__":
    import asyncio

    asyncio.run(resend_receipts_flow())