1. #!/bin/env python
    2. # encoding=utf-8
    4. import json
    5. import re
    6. import os
    7. import sys
    8. import ftplib
    9. import time
    10. import datetime
    11. import psycopg2
    12. from ftplib import FTP
    13. from kafka import KafkaProducer
    14. from multiprocessing.dummy import Pool as ThreadPool
    17. def conn_pg():
    18.     conn = psycopg2.connect(database='XX', user='XX', password='XXXX', host='XX.XX.XX.XX', port='XXXX')
    19.     print('Open database successfully')
    20.     return conn
    23. def conn_ftp(ftp_server, ftp_port, user_name, password):
    24.     try:
    25.         ftp = FTP()
    26.         ftp.connect(ftp_server, ftp_port)
    27.         ftp.login(user_name, password)
    28.     except Exception as e:
    29.         return None
    30.     else:
    31.         return ftp
    34. def ftp_scan(source_brokers, source_topic, infor):
    35.     ftp_server = infor[0]
    36.     ftp_port = int(infor[1])
    37.     ftp_name = infor[2]
    38.     ftp_pwd = infor[3]
    39.     remote_path = infor[5]
    40.     remote_bk_path = infor[6]
    41.     local_path = infor[7]
    42.     file_regular = infor[8]
    43.     chkf_regular = infor[9]
    44.     mv_flag = int(infor[10])
    45.     print(remote_path)
    47.     filepp=[]
    49.     ftp_conn = conn_ftp(ftp_server, ftp_port, ftp_name, ftp_pwd)
    50.     if not ftp_conn:
    51.         return filepp
    52.     ftp_conn.cwd(remote_path)
    53.     ftplist = ftp_conn.nlst()
    54.     # 遍历需要的数据文件、校验文件
    55.     for files in ftplist:
    56.         # 获取文件的时间
    57.         L = list(ftp_conn.sendcmd('MDTM ' + files))
    58.         # 将获取到的文件时间拼接
    59.         datetime_files = L[4] + L[5] + L[6] + L[7] + '-' + L[8] + L[9] + '-' + L[10] + L[11] + ' ' + L[12] + L[
    60.             13] + ':' + L[14] + L[15] + ':' + L[16] + L[17]
    61.         # 将str转换为datime格式
    62.         date_time_files = datetime.datetime.strptime(datetime_files, '%Y-%m-%d %H:%M:%S')
    63.         # utc时间+8小时，得到北京时间
    64.         sh_time = date_time_files + datetime.timedelta(hours=8)
    65.         file_size = ftp_conn.size(files)
    66.         if re.match(r"%s" % (file_regular), files) or re.match(r"%s" % (chkf_regular), files):
    69.            filepp.append(files)
    70.         jsonstr = '''filename":"{files}","filetime":"{sh_time}","filesize":"{file_size}"'''.format(
    71.             files=files, sh_time=sh_time, file_size=file_size)
    72.         print(jsonstr)
    73.         jsonstr = '{' + jsonstr.replace('\n', '') + '}'
    74.         kafka_producer(source_brokers, source_topic, jsonstr)
    76.     return filepp
    79. def file_down(file,infor):
    80.     #获取配置表相关信息
    81.     ftp_server = infor[0]
    82.     ftp_port = int(infor[1])
    83.     ftp_name = infor[2]
    84.     ftp_pwd = infor[3]
    85.     remote_path = infor[5]
    86.     remote_bk_path = infor[6]
    87.     local_path = infor[7]
    88.     ftp_conn = conn_ftp(ftp_server, ftp_port, ftp_name, ftp_pwd)
    89.     if ftp_conn:
    90.         print('downloading '+file)
    91.         filename = local_path + '/' + file
    92.         remotefile = remote_path + '/' + file
    93.         remotebkfile = remote_bk_path + '/' + file
    94.         try:
    95.             ftp_conn.cwd(remote_path)
    96.             f = open(filename, 'wb')
    97.             file_handle = f.write
    98.             ftp_conn.retrbinary("RETR %s" % (file), file_handle)
    99.             print('download complete '+file)
    100.             ftp_conn.rename(remotefile, remotebkfile)
    101.             print('rename complete '+file)
    102.             print('finsh '+file)
    103.         except Exception as e:
    104.             print(e)
    105.         ftp_conn.close()
    108. def kafka_producer(kafkabrokes, kafkatopic, jsonstr):
    109.     producer = KafkaProducer(bootstrap_servers=kafkabrokes)
    110.     try:
    111.         producer.send(kafkatopic, jsonstr)
    112.     except Exception as e:
    113.         print e
    114.     else:
    115.         producer.close()
    119. if __name__ == '__main__':
    121.     down_group_id = sys.argv[1]
    123.     source_brokers = ['XX.XX.XX.XX:XXXX', 'XX.XX.XX.XX:XXXX']   #brokes
    124.     source_topic = "XX"  # 批处理topic
    127.     sql = '''
    128.     SELECT
    129.     a.ip,
    130.     a.ftp_port,
    131.     a.ftp_name,
    132.     a.ftp_pwd,
    133.     a.home_path,
    134.     b.remote_path,
    135.     b.remote_bk_path,
    136.     b.local_path,
    137.     b.file_regular,
    138.     b.chkf_regular,
    139.     b.mv_flag,
    140.     b.data_source,
    141.     b.table_attr,
    142.     b.prov_id
    143.     FROM
    144.     public.conf_host_info a,
    145.     public.conf_path_info b
    146.     WHERE
    147.     a.host_key = b.host_key
    148.     AND b.is_valid = '0' and down_group_id = '{id}' and a.host_type = '3'
    149.     '''.format(id=down_group_id)
    151.     pg_conn = conn_pg()
    152.     cur = pg_conn.cursor()
    153.     cur.execute(sql)
    154.     rows = cur.fetchall()
    155.     print(rows)
    156.     pg_conn.close()
    158.     while True:
    160.         for i in rows:
    161.             files=ftp_scan(source_brokers, source_topic, i)
    162.             pool = ThreadPool(10)
    163.             # 为thread_q创建处理线程
    164.             for file in files:
    165.                 pool.apply_async(file_down, (file, i))
    167.             pool.close()
    168.             pool.join()
    170.         time.sleep(10)