首页下载资源后端file-download sftp

ZIPfile-download sftp

lavine081.29MB需要积分:1

资源文件列表:

file_download 大约有9个文件
  1. file_download/
  2. file_download/config.ini 1.29KB
  3. file_download/DEFECT.json 3.86KB
  4. file_download/file_download.py 55.03KB
  5. file_download/sftp需要安装的python依赖/
  6. file_download/sftp需要安装的python依赖/bcrypt-4.2.0-cp37-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl 267.08KB
  7. file_download/sftp需要安装的python依赖/paramiko-3.4.1-py3-none-any.whl 220.92KB
  8. file_download/sftp需要安装的python依赖/PyNaCl-1.5.0-cp36-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.manylinux_2_24_x86_64.whl 836.6KB
  9. file_download/t_biz_file_info.sql 6.14KB

资源介绍:

file-download sftp
#!/usr/bin/env python # encoding: utf-8 import argparse import configparser import ftplib import hashlib import json import logging import os import re import threading import time import traceback from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta from enum import Enum from pathlib import PurePosixPath from posixpath import dirname from queue import Queue, Empty import paramiko import pymysql logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class Enums(Enum): # 文件类型 PATH_TYPE_UNKNOWN = ("UNKOWN", 0) PATH_TYPE_FILE = ("FILE", 1) PATH_TYPE_DIR = ("DIR", 2) # 下载方式 DOWNLOAD_MODE_FULL = ("FULL", 0) DOWNLOAD_MODE_INCREMENT = ("INCRE", 1) # 文件操作 OPERATIONS_DEL = ("DEL", 0) OPERATIONS_MV = ("MV", 1) # 数据类型 DATA_TYPE_CP = ("CP", 2) DATA_TYPE_CP_SP = ("CP_SP", 20) DATA_TYPE_FT = ("FT", 3) DATA_TYPE_FT_SP = ("FT_SP", 30) DATA_TYPE_DEFECT = ("DEFECT", 5) # 文件状态 QUEUE = ("下载成功,在解析队列中", 1) TRASH_FILE = ("垃圾文件", 14) DOWNLOAD_FAIL = ("下载失败", 15) @property def DESC(self): return self._value_[0] @property def CODE(self): return self._value_[1] @classmethod def get_code_by_desc(cls, description): for member in cls: if str(member.DESC).strip().upper() == str(description).strip().upper(): return member.CODE raise ValueError(f"No member with description '{description}' found") class ConfigLoader: """ 加载配置文件内容 """ def __init__(self, config_file_path): self.config_file_path = config_file_path self.config = configparser.ConfigParser() self.config.read(config_file_path, encoding="UTF-8") self.starrocks_host = self.config.get("starrocks", 'host') self.starrocks_port = self.config.get("starrocks", 'port') self.starrocks_http_port = self.config.get("starrocks", 'http_port') self.starrocks_database = self.config.get("starrocks", 'database') self.starrocks_user = self.config.get("starrocks", 'user') self.starrocks_password = self.config.get("starrocks", 'password') self.mysql_host = self.config.get("mysql", 'host') self.mysql_port = self.config.get("mysql", 'port') self.mysql_database = self.config.get("mysql", 'database') self.mysql_user = self.config.get("mysql", 'user') self.mysql_password = self.config.get("mysql", 'password') self.encode_list = str(self.config.get("common", 'encode_list')).split(",") self.defect_white_list = str(self.config.get("common", 'defect_white_list')) self.defect_black_list = str(self.config.get("common", 'defect_black_list')) self.defect_trash_file_delete = str(self.config.get("common", 'defect_trash_file_delete')) self.pool_size = int(self.config.get("common", 'pool_size')) def get(self, section, option): try: value = self.config.get(section, option) return value except (configparser.NoSectionError, configparser.NoOptionError): return None def set_value(self, section, option, value): if not self.config.has_section(section): self.config.add_section(section) self.config.set(section, option, value) def save(self): with open(self.config_file_path, 'w') as configfile: self.config.write(configfile) class Tools: def generator_id(self, string_str): """ 计算字符串ID :param string_str: 字符串 :return: """ return abs(int.from_bytes(bytes.fromhex(hashlib.md5(str(string_str).encode()).hexdigest()[-16:]), 'little', signed=True)) def calculate_md5(self, file_abs_path): """ 计算文件 md5 :param file_abs_path: 文件绝对路径 :return: """ hash_md5 = hashlib.md5() with open(file_abs_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() def modify_file_name(self, lpath, md5): """ 修改所有文件名.可以根据需求自定义修改 :param lpath: :param md5: :return: """ file_name = os.path.basename(lpath) ldir = os.path.dirname(lpath) new_file_name = md5 + "::" + file_name new_lpath = os.path.join(ldir, new_file_name).replace("\\", '/') # 修改文件名 if os.path.exists(new_lpath): os.remove(new_lpath) os.rename(lpath, new_lpath) logging.info(f"Renamed: {lpath} -> {new_lpath}") return new_lpath def delete_file(self, file_abs_path): """ 判断指定路径的文件是否存在且是文件，若满足条件则删除该文件 :param file_abs_path: 要判断并删除的文件路径 :return: 无 """ if os.path.exists(file_abs_path) and os.path.isfile(file_abs_path): try: os.remove(file_abs_path) logging.warning(f"{file_abs_path}是垃圾文件, 已成功删除。") except FileNotFoundError: logging.info(f"{file_abs_path} 不存在，可能已被其他操作删除。") except PermissionError: logging.info(f"没有权限删除 {file_abs_path}。") else: logging.info(f"{file_abs_path} 不存在或不是一个文件。") def generator_file_list(self, backup_name, backup_path, bak_ftp_id, data_type, download_duration, download_end_time, download_start_time, download_status, file_list, instance_id, md5, mtime, new_lpath, rpath, size, src_ftp_id, config, tools): # 如果数据类型是cp_sp/ft_sp[cp_sp/ft_sp在一个工作流中,cp/sp数据都要录入] if str(data_type).strip().upper() == Enums.DATA_TYPE_CP_SP.DESC: # cp的录入一份 file_list.append([str(PurePosixPath(new_lpath).name), self.generator_id(str(PurePosixPath(rpath).name) + "::" + md5), Enums.DATA_TYPE_CP.CODE, 'data_source', mtime, size, md5, str(PurePosixPath(new_lpath).parent), str(PurePosixPath(rpath).parent), src_ftp_id, backup_name, backup_path, bak_ftp_id, download_duration, download_start_time, download_end_time, download_status, download_end_time, instance_id ]) # cp_sp的录入一份 file_list.append([str(PurePosixPath(new_lpath).name), self.generator_id(str(PurePosixPath(rpath).name) + "::" + md5), Enums.DATA_TYPE_CP_SP.CODE, 'data_source', mtime, size, md5, str(PurePosixPath(new_lpath).parent), str(PurePosixPath(rpath).parent), src_ftp_id, backup_name, backup_path, bak_ftp_id, download_duration, download_start_time,
100+评论
captcha

    相关资源

    1. MATLAB环境下多领域应用的高效变分模态分解算法优化研究：涵盖金融时间序列、地震微震信号、机械振动等多类信号的处理,MATLAB环境下多领域应用的高效变分模态分解算法优化研究：涵盖金融时间序列、地震

      bMHWiCiOKR#物联网ZIP685.76KB2周前
      资源摘要:MATLAB环境下多领域应用的高效变分模态分解算法优化研究：涵盖金融时间序列、地震微震信号、机械振动等多类信号的处理,MATLAB环境下多领域应用的高效变分模态分解算法优化研究：涵盖金融时间序列、地震微震信号、机械振动等多类信号的处理,MATLAB环境下一种改进的变分模态分解方法算法可迁移至金融时间序列，地震 微震信号，机械振动信号，声发射信号，电压 电流信号，语音信号，声信号，生理信号（ECG,EEG,EMG）等信号。,MATLAB环境; 改进变分模态分解方法; 算法可迁移性; 金融时间序列; 地震、微震信号; 机械振动信号; 声发射信号; 电压、电流信号; 语音信号; 生理信号,MATLAB中改进的变分模态分解算法：多领域信号处理新方法

    2. 异步电机矢量控制详解：FOC与转子磁场定向控制技术及其实践指南（附Word文档）,异步电机矢量控制与学习：聚焦FOC与转子磁场定向控制，高效控制效果及Word文档详解,异步电机矢量控制 foc 转

      bMHWiCiOKR#前端ZIP5.22MB2周前
      资源摘要:异步电机矢量控制详解：FOC与转子磁场定向控制技术及其实践指南（附Word文档）,异步电机矢量控制与学习：聚焦FOC与转子磁场定向控制，高效控制效果及Word文档详解,异步电机矢量控制 foc 转子磁场定向控制。是三套异步foc，可以对应学习，有相应的Word文档说明。控制效果很好。,异步电机; 矢量控制; FOC; 转子磁场定向控制; Word文档说明; 控制效果。,异步电机Foc矢量控制技术及其实施指南

    3. MATLAB仿真实现：Chan法计算TDOA与GDOP值的方法研究,MATLAB仿真实现：基于Chan法计算TDOA与GDOP算法研究,M00398-使用 Chan 法计算 TDOA 和计算 GDOP

      bMHWiCiOKR#移动开发ZIP2.1MB2周前
      资源摘要:MATLAB仿真实现：Chan法计算TDOA与GDOP值的方法研究,MATLAB仿真实现：基于Chan法计算TDOA与GDOP算法研究,M00398-使用 Chan 法计算 TDOA 和计算 GDOP 的 MATLAB 仿真实现,M00398;Chan法;TDOA计算;GDOP计算;MATLAB仿真实现,MATLAB仿真实现：TDOA与GDOP计算使用Chan法M00398

    4. 深入解析三线仿真问题解决方案,三线仿真问题的全面解析与解决方案,三线仿真问题解决,三线仿真; 问题解决; 解决方案; 故障排除,三线仿真难题攻破方案

      bMHWiCiOKR#行业研究ZIP912.78KB2周前
      资源摘要:深入解析三线仿真问题解决方案,三线仿真问题的全面解析与解决方案,三线仿真问题解决,三线仿真; 问题解决; 解决方案; 故障排除,三线仿真难题攻破方案

    5. 智能同步控制程序：S7-200 Smart在卷板材生产线与造纸设备的应用,多机同步控制与速度频率同步程序-专为卷板材生产线和造纸设备设计,速度同步频率同步程序s7-200smart程序适合卷板材

      bMHWiCiOKR#操作系统ZIP340.83KB2周前
      资源摘要:智能同步控制程序：S7-200 Smart在卷板材生产线与造纸设备的应用,多机同步控制与速度频率同步程序——专为卷板材生产线和造纸设备设计,速度同步频率同步程序s7-200smart程序适合卷板材生产线，造纸设备主同步控制程序适合英威腾。ABB变频器，国产变频器适合多机同步控制的速度频率同步程序。只有同步频率部分，速度同步地址送到比如vw10,带同步速度跟随，主从单机微调。不包含通讯启停，通许写频率，读电流部分。由于软件具有可复制性。，提供变量表，数据说明，1-15回路内同步频率部分。1-30回路需要备注说明。,速度同步;频率同步;s7-200smart程序;卷板材生产线同步控制;主同步控制程序;变频器;国产变频器;同步速度跟随;微调;回路同步频率;变量表;数据说明,自动化控制：速度与频率同步程序方案

    6. 六电池智能均衡系统：高精度、快速均衡的Buck-Boost电路解决方案,高效电池均衡技术：快速精准调节BuckBoost电路，提升电池性能与寿命,6个电池均衡，buckboost电路，精度高，均衡速度

      bMHWiCiOKR#信息化管理ZIP1008.38KB2周前
      资源摘要:六电池智能均衡系统：高精度、快速均衡的Buck-Boost电路解决方案,高效电池均衡技术：快速精准调节BuckBoost电路，提升电池性能与寿命,6个电池均衡，buckboost电路，精度高，均衡速度快,电池均衡; buck-boost电路; 精度高; 均衡速度快; 充电技术; 能量回收。,六电池快速均衡系统：高精度Buck-Boost电路技术

    7. MATLAB程序实现：多无人船协同围捕控制算法-3船围捕运动船只仿真，距离监控与学习参考,MATLAB程序实现：多无人船协同围捕控制算法-3船围捕运动船只仿真与距离动态调整,MATLAB程序：多个

      bMHWiCiOKR#人工智能ZIP162.34KB2周前
      资源摘要:MATLAB程序实现：多无人船协同围捕控制算法——3船围捕运动船只仿真，距离监控与学习参考,MATLAB程序实现：多无人船协同围捕控制算法——3船围捕运动船只仿真与距离动态调整,MATLAB程序：多个无人船 协同围捕控制算法3船围捕控制，围捕运动船只可以仿真多个船之间的距离以及距离目标船的距离，特别适合学习、参考,MATLAB程序; 无人船; 协同围捕控制算法; 3船围捕控制; 距离控制; 仿真; 学习、参考,MATLAB算法：多船协同围捕控制，三船动态包围策略模拟学习参考

    8. 双极性SPWM单相全桥逆变电路仿真模型：电压电流双闭环控制，直流输入电压范围广泛，输出交流峰值电压可调，高效频率控制技术在1-200hz之间 ,双极性SPWM单相全桥逆变电路仿真模型：电压电流双闭环控

      bMHWiCiOKR#前端ZIP302.86KB2周前
      资源摘要:双极性SPWM单相全桥逆变电路仿真模型：电压电流双闭环控制，直流输入电压范围广泛，输出交流峰值电压可调，高效频率控制技术在1-200hz之间。,双极性SPWM单相全桥逆变电路仿真模型：电压电流双闭环控制，直流输入与交变输出性能探究,双极性SPWM控制单相全桥逆变电路仿真模型，电压电流双闭环控制。直流输入电压范围在10-40v左右，输出交流峰值在正负10-40v，频率1-200hz可调。,双极性SPWM控制; 单相全桥逆变电路仿真模型; 电压电流双闭环控制; 直流输入电压范围; 输出交流峰值; 频率可调。,双极性SPWM控制全桥逆变电路仿真：电压电流双闭环调节与频率可调
    站长QQ:九六九五七三七