scripts/flow_backup.py - master - beyondviolet/edw_prefect

scripts/flow_backup.py
78 строк2 KB
get458
first_commit
04 авг 2026, 07:38
04 авг 2026, 07:38b3e3bb3
from prefect import flow, task
import asyncio
import logging
import pandas as pd
from bvcore.db import DBPool
from bvcore.app import set_cwd

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'))
bv_log.addHandler(handler)

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


@task
def fetch_receipts(limit: int = 45):
    async def _fetch():
        await db.init_pool()
        pool = db('default')
        sql_query = """
            SELECT pr.uuid, pr.external_code, pr.status, pg.code as gateway_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 = 'canceled'
              AND resend_count < 5
            LIMIT $1
        """
        async with pool.acquire() as conn:
            rows = await conn.fetch(sql_query, limit)
        return [dict(r) for r in rows]

    return asyncio.run(_fetch())

@task
def reset_receipt(rcpt_uuid: str):
    async def _reset():
        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
            WHERE uuid = $1
        """
        async with pool.acquire() as conn:
            await conn.execute(sql, rcpt_uuid)
        bv_log.info(f"Receipt {rcpt_uuid} reset to pending.")

    return asyncio.run(_reset())


@flow(name="Resend Canceled Receipts")
def resend_receipts_flow(limit: int = 45):
    receipts = fetch_receipts(limit=limit)
    processed_results = []

    for rcpt in receipts:
        reset_receipt.submit(rcpt["uuid"]).result()

    df = pd.DataFrame(processed_results)
    bv_log.info(f"Flow finished. Resends: {len(df)}")
    print(df)


if __name__ == "__main__":
    resend_receipts_flow()