util ftp renamed connect
This commit is contained in:
367
src/utils/connect/send_data.py
Normal file
367
src/utils/connect/send_data.py
Normal file
@@ -0,0 +1,367 @@
|
||||
from ftplib import FTP, FTP_TLS, all_errors
|
||||
from io import BytesIO
|
||||
import logging
|
||||
import aiomysql
|
||||
from datetime import datetime
|
||||
|
||||
from utils.database.loader_action import update_status, unlock
|
||||
from utils.database.action_query import get_data_as_csv, get_tool_info, get_elab_timestamp
|
||||
from utils.database import WorkflowFlags
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class FTPConnection:
|
||||
"""
|
||||
Manages an FTP or FTP_TLS connection, providing a context manager for automatic disconnection.
|
||||
"""
|
||||
def __init__(self, host, port=21, use_tls=False, user='', passwd='',
|
||||
passive=True, timeout=None, debug=0, context=None):
|
||||
|
||||
self.use_tls = use_tls
|
||||
|
||||
if use_tls:
|
||||
self.ftp = FTP_TLS(context=context, timeout=timeout) if context else FTP_TLS(timeout=timeout)
|
||||
else:
|
||||
self.ftp = FTP(timeout=timeout)
|
||||
|
||||
if debug > 0:
|
||||
self.ftp.set_debuglevel(debug)
|
||||
|
||||
self.ftp.connect(host, port)
|
||||
self.ftp.login(user, passwd)
|
||||
self.ftp.set_pasv(passive)
|
||||
|
||||
if use_tls:
|
||||
self.ftp.prot_p()
|
||||
|
||||
def __getattr__(self, name):
|
||||
"""Delega tutti i metodi non definiti all'oggetto FTP sottostante"""
|
||||
return getattr(self.ftp, name)
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
self.ftp.quit()
|
||||
|
||||
async def ftp_send_raw_csv_to_customer(cfg: dict, id: int, unit: str, tool: str, pool: object) -> bool:
|
||||
None
|
||||
return True
|
||||
|
||||
async def ftp_send_elab_csv_to_customer(cfg: dict, id: int, unit: str, tool: str, csv_data: str, pool: object) -> bool:
|
||||
"""
|
||||
Sends elaborated CSV data to a customer via FTP.
|
||||
|
||||
Retrieves FTP connection details from the database based on the unit name,
|
||||
then establishes an FTP connection and uploads the CSV data.
|
||||
|
||||
Args:
|
||||
cfg (dict): Configuration dictionary (not directly used in this function but passed for consistency).
|
||||
id (int): The ID of the record being processed (used for logging).
|
||||
unit (str): The name of the unit associated with the data.
|
||||
tool (str): The name of the tool associated with the data.
|
||||
csv_data (str): The CSV data as a string to be sent.
|
||||
pool (object): The database connection pool.
|
||||
|
||||
Returns:
|
||||
bool: True if the CSV data was sent successfully, False otherwise.
|
||||
"""
|
||||
query = """
|
||||
select ftp_addrs, ftp_user, ftp_passwd, ftp_parm, ftp_filename, ftp_target, duedate from units
|
||||
where name = '%s'";'
|
||||
"""
|
||||
async with pool.acquire() as conn:
|
||||
async with conn.cursor(aiomysql.DictCursor) as cur:
|
||||
try:
|
||||
await cur.execute(query, (unit,))
|
||||
send_ftp_info = await cur.fetchone()
|
||||
logger.info(f"id {id} - {unit} - {tool}: estratti i dati per invio via ftp")
|
||||
except Exception as e:
|
||||
logger.error(f"id {id} - {unit} - {tool} - errore nella query per invio ftp: {e}")
|
||||
|
||||
try:
|
||||
# Converti in bytes
|
||||
csv_bytes = csv_data.encode('utf-8')
|
||||
csv_buffer = BytesIO(csv_bytes)
|
||||
|
||||
ftp_parms = await parse_ftp_parms(send_ftp_info["ftp_parm"])
|
||||
use_tls = 'ssl_version' in ftp_parms
|
||||
passive = ftp_parms.get('passive', True)
|
||||
port = ftp_parms.get('port', 21)
|
||||
|
||||
# Connessione FTP
|
||||
with FTPConnection(host=send_ftp_info["ftp_addrs"], port=port, use_tls=use_tls, user=send_ftp_info["ftp_user"], passwd=send_ftp_info["ftp_passwd"], passive=passive) as ftp:
|
||||
|
||||
# Cambia directory
|
||||
if send_ftp_info["ftp_target"] != "/":
|
||||
ftp.cwd(send_ftp_info["ftp_target"])
|
||||
|
||||
# Invia il file
|
||||
result = ftp.storbinary(f'STOR {send_ftp_info["ftp_filename"]}', csv_buffer)
|
||||
|
||||
if result.startswith('226'):
|
||||
logger.info(f"File {send_ftp_info["ftp_filename"]} inviato con successo")
|
||||
return True
|
||||
else:
|
||||
logger.error(f"Errore nell'invio: {result}")
|
||||
return False
|
||||
|
||||
except all_errors as e:
|
||||
logger.error(f"Errore FTP: {e}")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"Errore generico: {e}")
|
||||
return False
|
||||
finally:
|
||||
csv_buffer.close()
|
||||
|
||||
async def parse_ftp_parms(ftp_parms: str) -> dict:
|
||||
"""
|
||||
Parses a string of FTP parameters into a dictionary.
|
||||
|
||||
Args:
|
||||
ftp_parms (str): A string containing key-value pairs separated by commas,
|
||||
with keys and values separated by '=>'.
|
||||
|
||||
Returns:
|
||||
dict: A dictionary where keys are parameter names (lowercase) and values are their parsed values.
|
||||
"""
|
||||
# Rimuovere spazi e dividere per virgola
|
||||
pairs = ftp_parms.split(',')
|
||||
result = {}
|
||||
|
||||
for pair in pairs:
|
||||
if '=>' in pair:
|
||||
key, value = pair.split('=>', 1)
|
||||
key = key.strip().lower()
|
||||
value = value.strip().lower()
|
||||
|
||||
# Convertire i valori appropriati
|
||||
if value.isdigit():
|
||||
value = int(value)
|
||||
elif value == '':
|
||||
value = None
|
||||
|
||||
result[key] = value
|
||||
|
||||
return result
|
||||
|
||||
|
||||
async def process_workflow_record(record: tuple, fase: int, cfg: dict, pool: object):
|
||||
"""
|
||||
Elabora un singolo record del workflow in base alla fase specificata.
|
||||
|
||||
Args:
|
||||
record: Tupla contenente i dati del record
|
||||
fase: Fase corrente del workflow
|
||||
cfg: Configurazione
|
||||
pool: Pool di connessioni al database
|
||||
"""
|
||||
# Estrazione e normalizzazione dei dati del record
|
||||
id, unit_type, tool_type, unit_name, tool_name = [
|
||||
x.lower().replace(" ", "_") if isinstance(x, str) else x
|
||||
for x in record
|
||||
]
|
||||
|
||||
try:
|
||||
# Recupero informazioni principali
|
||||
tool_elab_info = await get_tool_info(fase, unit_name.upper(), tool_name.upper(), pool)
|
||||
if tool_elab_info:
|
||||
timestamp_matlab_elab = await get_elab_timestamp(id, pool)
|
||||
|
||||
# Verifica se il processing può essere eseguito
|
||||
if not _should_process(tool_elab_info, timestamp_matlab_elab):
|
||||
logger.info(f"id {id} - {unit_name} - {tool_name} {tool_elab_info['duedate']}: "
|
||||
"invio dati non eseguito - due date raggiunta.")
|
||||
|
||||
await update_status(cfg, id, fase, pool)
|
||||
return
|
||||
|
||||
# Routing basato sulla fase
|
||||
success = await _route_by_phase(fase, tool_elab_info, cfg, id, unit_name, tool_name,
|
||||
timestamp_matlab_elab, pool)
|
||||
|
||||
if success:
|
||||
await update_status(cfg, id, fase, pool)
|
||||
else:
|
||||
await update_status(cfg, id, fase, pool)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Errore durante elaborazione id {id} - {unit_name} - {tool_name}: {e}")
|
||||
raise
|
||||
finally:
|
||||
await unlock(cfg, id, pool)
|
||||
|
||||
|
||||
def _should_process_old(tool_elab_info, timestamp_matlab_elab):
|
||||
"""Verifica se il record può essere processato basandosi sulla due date."""
|
||||
duedate = tool_elab_info.get("duedate")
|
||||
|
||||
if not duedate or duedate in ('0000-00-00 00:00:00', ''):
|
||||
return True
|
||||
|
||||
# Se timestamp_matlab_elab è None/null, usa il timestamp corrente
|
||||
comparison_timestamp = timestamp_matlab_elab if timestamp_matlab_elab is not None else datetime.now()
|
||||
|
||||
return duedate > comparison_timestamp
|
||||
|
||||
|
||||
def _should_process(tool_elab_info, timestamp_matlab_elab):
|
||||
"""Verifica se il record può essere processato basandosi sulla due date."""
|
||||
duedate = tool_elab_info.get("duedate")
|
||||
|
||||
# Se non c'è duedate o è vuota/nulla, può essere processato
|
||||
if not duedate or duedate in ('0000-00-00 00:00:00', ''):
|
||||
return True
|
||||
|
||||
# Se timestamp_matlab_elab è None/null, usa il timestamp corrente
|
||||
comparison_timestamp = timestamp_matlab_elab if timestamp_matlab_elab is not None else datetime.now()
|
||||
|
||||
# Converti duedate in datetime se è una stringa
|
||||
if isinstance(duedate, str):
|
||||
duedate = datetime.strptime(duedate, '%Y-%m-%d %H:%M:%S')
|
||||
|
||||
# Assicurati che comparison_timestamp sia datetime
|
||||
if isinstance(comparison_timestamp, str):
|
||||
comparison_timestamp = datetime.strptime(comparison_timestamp, '%Y-%m-%d %H:%M:%S')
|
||||
|
||||
return duedate > comparison_timestamp
|
||||
|
||||
|
||||
|
||||
async def _route_by_phase(fase, tool_elab_info, cfg, id, unit_name, tool_name,
|
||||
timestamp_matlab_elab, pool):
|
||||
"""
|
||||
Gestisce il routing delle operazioni in base alla fase del workflow.
|
||||
|
||||
Returns:
|
||||
bool: True se l'operazione è riuscita, False altrimenti
|
||||
"""
|
||||
if fase == WorkflowFlags.SENT_ELAB_DATA:
|
||||
return await _handle_elab_data_phase(tool_elab_info, cfg, id, unit_name,
|
||||
tool_name, timestamp_matlab_elab, pool)
|
||||
|
||||
elif fase == WorkflowFlags.SENT_RAW_DATA:
|
||||
return await _handle_raw_data_phase(tool_elab_info, cfg, id, unit_name,
|
||||
tool_name, pool)
|
||||
|
||||
else:
|
||||
logger.info(f"id {id} - {unit_name} - {tool_name}: nessuna azione da eseguire.")
|
||||
return True
|
||||
|
||||
|
||||
async def _handle_elab_data_phase(tool_elab_info, cfg, id, unit_name, tool_name,
|
||||
timestamp_matlab_elab, pool):
|
||||
"""Gestisce la fase di invio dati elaborati."""
|
||||
|
||||
# FTP send per dati elaborati
|
||||
if tool_elab_info.get('ftp_send'):
|
||||
return await _send_elab_data_ftp(cfg, id, unit_name, tool_name,
|
||||
timestamp_matlab_elab, pool)
|
||||
|
||||
# API send per dati elaborati
|
||||
elif _should_send_elab_api(tool_elab_info):
|
||||
return await _send_elab_data_api(cfg, id, unit_name, tool_name,
|
||||
timestamp_matlab_elab, pool)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def _handle_raw_data_phase(tool_elab_info, cfg, id, unit_name, tool_name, pool):
|
||||
"""Gestisce la fase di invio dati raw."""
|
||||
|
||||
# FTP send per dati raw
|
||||
if tool_elab_info.get('ftp_send_raw'):
|
||||
return await _send_raw_data_ftp(cfg, id, unit_name, tool_name, pool)
|
||||
|
||||
# API send per dati raw
|
||||
elif _should_send_raw_api(tool_elab_info):
|
||||
return await _send_raw_data_api(cfg, id, unit_name, tool_name, pool)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def _should_send_elab_api(tool_elab_info):
|
||||
"""Verifica se i dati elaborati devono essere inviati via API."""
|
||||
return (tool_elab_info.get('inoltro_api') and
|
||||
tool_elab_info.get('api_send') and
|
||||
tool_elab_info.get('inoltro_api_url', '').strip())
|
||||
|
||||
|
||||
def _should_send_raw_api(tool_elab_info):
|
||||
"""Verifica se i dati raw devono essere inviati via API."""
|
||||
return (tool_elab_info.get('inoltro_api_raw') and
|
||||
tool_elab_info.get('api_send_raw') and
|
||||
tool_elab_info.get('inoltro_api_url_raw', '').strip())
|
||||
|
||||
|
||||
async def _send_elab_data_ftp(cfg, id, unit_name, tool_name, timestamp_matlab_elab, pool):
|
||||
"""Invia dati elaborati via FTP."""
|
||||
try:
|
||||
elab_csv = await get_data_as_csv(cfg, id, unit_name, tool_name,
|
||||
timestamp_matlab_elab, pool)
|
||||
if not elab_csv:
|
||||
return False
|
||||
|
||||
print(elab_csv)
|
||||
# if await send_elab_csv_to_customer(cfg, id, unit_name, tool_name, elab_csv, pool):
|
||||
if True: # Placeholder per test
|
||||
return True
|
||||
else:
|
||||
logger.error(f"id {id} - {unit_name} - {tool_name}: invio FTP fallito.")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Errore invio FTP elab data id {id}: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def _send_elab_data_api(cfg, id, unit_name, tool_name, timestamp_matlab_elab, pool):
|
||||
"""Invia dati elaborati via API."""
|
||||
try:
|
||||
elab_csv = await get_data_as_csv(cfg, id, unit_name, tool_name,
|
||||
timestamp_matlab_elab, pool)
|
||||
if not elab_csv:
|
||||
return False
|
||||
|
||||
print(elab_csv)
|
||||
# if await send_elab_csv_to_customer(cfg, id, unit_name, tool_name, elab_csv, pool):
|
||||
if True: # Placeholder per test
|
||||
return True
|
||||
else:
|
||||
logger.error(f"id {id} - {unit_name} - {tool_name}: invio API fallito.")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Errore invio API elab data id {id}: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def _send_raw_data_ftp(cfg, id, unit_name, tool_name, pool):
|
||||
"""Invia dati raw via FTP."""
|
||||
try:
|
||||
# if await ftp_send_raw_csv_to_customer(cfg, id, unit_name, tool_name, pool):
|
||||
if True: # Placeholder per test
|
||||
return True
|
||||
else:
|
||||
logger.error(f"id {id} - {unit_name} - {tool_name}: invio FTP raw fallito.")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Errore invio FTP raw data id {id}: {e}")
|
||||
return False
|
||||
|
||||
|
||||
async def _send_raw_data_api(cfg, id, unit_name, tool_name, pool):
|
||||
"""Invia dati raw via API."""
|
||||
try:
|
||||
# if await api_send_raw_csv_to_customer(cfg, id, unit_name, tool_name, pool):
|
||||
if True: # Placeholder per test
|
||||
return True
|
||||
else:
|
||||
logger.error(f"id {id} - {unit_name} - {tool_name}: invio API raw fallito.")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Errore invio API raw data id {id}: {e}")
|
||||
return False
|
||||
Reference in New Issue
Block a user