### 实现 Python 脚本从 FTP 服务器拉取文件并存储至腾讯云 COS 以下是一个完整的解决方案，涵盖了从 FTP 服务器拉取文件、去重、验证文件完整性以及将文件存储到腾讯云 COS 的功能。 #### 1. 拉取 FTP 文件 使用 `ftplib` 库可以从 FTP 服务器下载文件。为了确保高效性，可以通过 MD5 或 SHA-256 哈希算法计算文件指纹以检测重复文件。 ```python from ftplib import FTP import hashlib import os def fetch_file_from_ftp(ftp_server, username, password, remote_path, local_path): try: ftp = FTP(ftp_server) ftp.login(user=username, passwd=password) with open(local_path, 'wb') as f: ftp.retrbinary(f'RETR {remote_path}', f.write) ftp.quit() # 计算文件哈希值用于后续去重 hash_md5 = hashlib.md5() with open(local_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) return hash_md5.hexdigest() # 返回文件的MD5值作为唯一标识符 except Exception as e: print(f"FTP操作失败: {e}") return None ``` #### 2. 去重逻辑 通过维护一个记录已处理文件哈希值的集合（如 Redis 数据库或本地文件），可以轻松实现去重功能。 ```python def is_duplicate(file_hash, duplicate_set="processed_files.txt"): if not os.path.exists(duplicate_set): open(duplicate_set, 'w').close() with open(duplicate_set, 'r+') as f: hashes = set(line.strip() for line in f.readlines()) if file_hash in hashes: return True else: f.seek(0, 2) # 移动指针到最后 f.write(f"{file_hash}\n") return False ``` #### 3. 验证文件完整性 在上传之前，再次校验文件的哈希值是否一致，确保文件未被篡改。 ```python def verify_integrity(local_path, expected_hash): hash_md5 = hashlib.md5() with open(local_path, "rb") as f: for chunk in iter(lambda: f.read(4096), b""): hash_md5.update(chunk) actual_hash = hash_md5.hexdigest() return actual_hash == expected_hash ``` #### 4. 将文件存储到腾讯云 COS 借助腾讯云官方 SDK (`cos-python-sdk-v5`) 可以方便地将文件上传到 COS。 ```python from qcloud_cos import CosConfig, CosS3Client def upload_to_cos(secret_id, secret_key, region, bucket_name, local_path, cos_object_name): config = CosConfig( SecretId=secret_id, SecretKey=secret_key, Region=region ) client = CosS3Client(config) response = client.upload_file( Bucket=bucket_name, LocalFilePath=local_path, Key=cos_object_name ) return response ``` #### 综合脚本流程 将以上模块组合成一个完整的脚本： ```python if __name__ == "__main__": ftp_server = "ftp.example.com" username = "your_username" password = "your_password" remote_path = "/path/to/remote/file.txt" local_path = "./downloaded_file.txt" secret_id = "your_secret_id" secret_key = "your_secret_key" region = "ap-beijing" bucket_name = "example-bucket-1234567890" cos_object_name = "uploaded_file.txt" # Step 1: Fetch the file from FTP server and calculate its hash. file_hash = fetch_file_from_ftp(ftp_server, username, password, remote_path, local_path)[^1] if not file_hash: exit("Failed to retrieve file.") # Step 2: Check for duplicates based on the calculated hash. if is_duplicate(file_hash)[^2]: print("File already exists; skipping...") else: # Step 3: Verify integrity of downloaded file. if verify_integrity(local_path, file_hash)[^3]: # Step 4: Upload verified file to Tencent Cloud COS. result = upload_to_cos(secret_id, secret_key, region, bucket_name, local_path, cos_object_name)[^4] print(f"Upload successful! Response: {result}") else: print("Integrity check failed.") ``` --- ###