chip

chip1stop.py

+import ftplib
+import gzip
+import os
+
+import pika
+
+host = "ansetchip1.gotoftp11.com"
+user = "ansetchip1"
+password = "nf6l2g17"
+port = 21
+ftp_remote_path = "/Chip1Stop_TI_included_20240826_ETCHIPS.csv.gz"
+file_local_path = "./Chip1Stop_TI_included_20240826_ETCHIPS.csv.gz"
+
+ENV = "test"  # ENV 取值 test 或 prod
+
+config = {}
+
+config_test = {
+    "rabbit_mq_host": "192.168.1.237",
+    "rabbit_mq_queue_producer": "europa_erp_sku_queue",
+    "rabbit_mq_port": 5672,
+    "rabbit_mq_user": "huntadmin",
+    "rabbit_mq_password": "jy2y2900",
+    "rabbit_mq_exchange": "europa_erp_parse_file_exchange",
+    "rabbit_mq_routing_key": "europa_erp_sku_routing",
+}
+
+config_prod = {
+    "rabbit_mq_host": "119.23.79.136",
+    "rabbit_mq_queue_producer": "europa_erp_sku_queue",
+    "rabbit_mq_port": 5672,
+    "rabbit_mq_user": "ans2024",
+    "rabbit_mq_password": "ans2024123",
+    "rabbit_mq_exchange": "europa_erp_parse_file_exchange",
+    "rabbit_mq_routing_key": "europa_erp_sku_routing",
+}
+
+
+class DataUtil:
+
+    def __init__(self):
+        ftp = ftplib.FTP()
+        ftp.connect(host, port)
+        ftp.login(user, password)
+        ftp.set_pasv(False)
+        self.ftp = ftp
+
+    def download(self, remote_path, local_path):
+        fp = open(local_path, "wb")
+        self.ftp.retrbinary('RETR ' + remote_path, fp.write)
+        fp.close()
+
+    def ungz(self, file_path: str):
+        """解压缩gz格式文件"""
+        f_name = file_path.replace(".gz", "")
+        g_file = gzip.GzipFile(file_path)
+        open(f_name, "wb+").write(g_file.read())
+        g_file.close()
+        os.remove(file_path)
+
+    def handle_csv(self):
+        pass
+
+    def handle_txt(self):
+        pass
+
+
+class Producer:
+    def __init__(self):
+        credentials = pika.PlainCredentials(username=config["rabbit_mq_host"], password=config["rabbit_mq_password"])
+        self.conn = pika.BlockingConnection(pika.ConnectionParameters(
+            host=config["rabbit_mq_host"],
+            port=config["rabbit_mq_port"],
+            credentials=credentials
+        ))
+        channel = self.conn.channel()
+        channel.queue_declare(queue=config["rabbit_mq_queue_producer"], durable=True)
+        self.channel = channel
+
+    def push(self, message):
+        self.channel.basic_publish(
+            exchange=config["rabbit_mq_exchange"],
+            routing_key=config["rabbit_mq_routing_key"],
+            body=str.encode(message)
+        )
+
+    def close(self):
+        try:
+            self.channel.close()
+            self.conn.close()
+        except Exception as e:
+            print(e)
+
+
+if __name__ == '__main__':
+    if ENV == "test":
+        config = config_test
+    elif ENV == "prod":
+        config = config_prod
+    u = DataUtil()
+    u.download(ftp_remote_path, file_local_path)
+    u.ungz(file_local_path)