53 lines
2.2 KiB
Python
53 lines
2.2 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
# @Time : 2025/3/18 16:31
|
|
# @Author : AngesZhu
|
|
# @File : move_excel_service.py
|
|
# @Desc : 复制并移动目标文件至渠道数据
|
|
import os
|
|
import shutil
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from utils.logger_utils import logger
|
|
from utils.path_utils import PathOperator
|
|
|
|
|
|
def copy_and_rename_file(source_path, env: str, **kwargs):
|
|
"""
|
|
复制文件到目标目录并更新文件名。
|
|
:param source_path: 源文件的完整路径（包括文件名）
|
|
"""
|
|
try:
|
|
path_operator = PathOperator()
|
|
temp_file_path = path_operator.get_parent_path(source_path)
|
|
file_path = path_operator.get_parent_path(temp_file_path)
|
|
target_path = os.path.join(file_path, "channel_data")
|
|
logger.info(f"拼接后的复制目录{target_path}")
|
|
# 确保目标目录存在，如果不存在则创建
|
|
os.makedirs(target_path, exist_ok=True)
|
|
# 生成新的文件名
|
|
name, ext = os.path.splitext(source_path.split('/')[-1])
|
|
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
if kwargs and 'business_line' in kwargs:
|
|
kwargs_str = f'business_line_{kwargs["business_line"]}'
|
|
# kwargs_list = [f"{key}_{value}" for key, value in kwargs.items()]
|
|
# kwargs_str = '_'.join(kwargs_list)
|
|
new_filename = f"{env}_{name}_{kwargs_str}_{timestamp}{ext}" # 新的文件名
|
|
else:
|
|
new_filename = f"{env}_{name}_{timestamp}{ext}"
|
|
# new_filename = f"{env}_{name}_{timestamp}{ext}"
|
|
# 构造目标文件的完整路径
|
|
destination_path = os.path.join(target_path, new_filename)
|
|
logger.info(f"构造目标文件的完整路径{destination_path}")
|
|
# 使用 shutil.copy 复制文件
|
|
shutil.copy(source_path, destination_path)
|
|
logger.info(f"文件已成功复制到: {destination_path}")
|
|
return destination_path
|
|
except FileNotFoundError:
|
|
logger.error(f"错误：源文件 '{source_path}' 不存在。")
|
|
except PermissionError:
|
|
logger.error(f"错误：没有权限访问文件或目录。")
|
|
except Exception as e:
|
|
logger.error(f"发生错误: {e}")
|