python 实现保存最新的三份文件,其余的都删掉

yipeiwu_com5年前Python基础

我就废话不多说了,直接上代码吧!

"""
对于每天存储文件,文件数量过多,占用空间
采用保存最新的三个文件
"""
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.models import Variable
from sctetl.airflow.utils import dateutils
from datetime import datetime,timedelta
import logging
import os
import shutil
"""
base_dir = "/data"
data_dir = "/gather"
"gather下边存在不同的文件夹"
"/data/gather/test"
"test路径下有以下文件夹"
"20180812、20180813、20180814、20180815、20180816"
"""
 
base_dir = Variable.get("base_dir")
data_dir = Variable.get("data_dir")
keep = 3
 
default_arg = {
  "owner":"airflow",
  "depends_on_past":False,
  "start_date":dateutils.get_start_date_local(2018,8,27,18,5),
  "email":[''],
  "email_on_failure":False,
  "email_on_retry":False,
  "retries":1,
  "retry_delay":timedelta(minutes=5)
}
 
dag = DAG(dag_id="keep_three_day",default_args=default_arg,schedule_interval=dateutils.get_schedule_interval_local(18,5))
 
def keep_three_day():
  path = os.path.join(base_dir, data_dir)
  date_cates = os.listdir(path)
  for cate in date_cates:
    p = os.path.join(base_dir, data_dir, cate)
    if os.path.isdir(p):
      dir_names = os.listdir(p)
      dir_names.sort()
      for i in dir_names[:-keep]:
        logging.info("删除目录 {path}".format(path=os.path.join(p, i)))
        shutil.rmtree(os.path.join(p, i))
 
with dag:
  keep_three_file = PythonOperator(task_id="keep_three_file",python_callable=keep_three_day(),dag=dag)
 
keep_three_file
 

以上这篇python 实现保存最新的三份文件,其余的都删掉就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持【听图阁-专注于Python设计】。

相关文章

关于阿里云oss获取sts凭证 app直传 python的实例

首先安装需要的模块 pip install aliyun-python-sdk-core pip install aliyun-python-sdk-sts 获取需要的5个参数 g...

pyqt5中QThread在使用时出现重复emit的实例

在PyQt5中使用QThread的时候,要注意把所有QThread的对象在主类中的init(或者放在所有类函数的外面)中进行实例化,不然可能在多个QThread互相调用的时候,emit重...

Python wxpython模块响应鼠标拖动事件操作示例

Python wxpython模块响应鼠标拖动事件操作示例

本文实例讲述了Python wxpython模块响应鼠标拖动事件操作。分享给大家供大家参考,具体如下: wxpython鼠标拖动事件小案例: #coding:UTF-8 import...

python3.4下django集成使用xadmin后台的方法

python3.4下django集成使用xadmin后台的方法

环境:window7 x64、python3.4、django1.10 一、pip install xadmin安装报错 1、使用pip install xadmin命令安装可能报如下错...

python 调试冷知识(小结)

python 调试冷知识(小结)

对于 python 代码的调试我们通常都是使用 IDE 自带的调试功能。但是 IDE 提供的调试功能存在局限性,例如在测试服务器上调试代码,但是又不可能在测试服务器上安装 IDE 进行调...