fix: resolve all Ruff linting errors
- Fix PLW2901: Use different variable name for stripped lines in loop - Fix E721: Use 'is' instead of '==' for type comparisons - Add noqa comments for intentional security warnings (S321, S507, S110, S603) Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -57,7 +57,7 @@ def upload_to_ftp(self, file_path: str, file_id: int = None):
|
||||
if use_tls:
|
||||
try:
|
||||
logger.info(f"Attempting FTPS connection to {settings.ftp_host}")
|
||||
ftp = ftplib.FTP_TLS()
|
||||
ftp = ftplib.FTP_TLS() # noqa: S321
|
||||
ftp.connect(host=settings.ftp_host, port=settings.ftp_port or 21)
|
||||
|
||||
# Login with credentials
|
||||
@@ -74,7 +74,7 @@ def upload_to_ftp(self, file_path: str, file_id: int = None):
|
||||
else:
|
||||
logger.warning(f"FTPS connection failed, falling back to regular FTP: {str(e)}")
|
||||
# Fall back to regular FTP - only if explicitly allowed by configuration
|
||||
ftp = ftplib.FTP() # nosec B321 - Fallback to FTP intentional when configured
|
||||
ftp = ftplib.FTP() # nosec B321 - Fallback to FTP intentional when configured # noqa: S321
|
||||
ftp.connect(host=settings.ftp_host, port=settings.ftp_port or 21)
|
||||
|
||||
# Login with credentials
|
||||
@@ -88,7 +88,7 @@ def upload_to_ftp(self, file_path: str, file_id: int = None):
|
||||
|
||||
# Directly use regular FTP if TLS is explicitly disabled
|
||||
logger.warning("Using plaintext FTP - connection is NOT encrypted!")
|
||||
ftp = ftplib.FTP() # nosec B321 - Plaintext FTP intentional when explicitly configured
|
||||
ftp = ftplib.FTP() # nosec B321 - Plaintext FTP intentional when explicitly configured # noqa: S321
|
||||
ftp.connect(host=settings.ftp_host, port=settings.ftp_port or 21)
|
||||
|
||||
# Login with credentials
|
||||
|
||||
@@ -56,7 +56,7 @@ def upload_to_sftp(self, file_path: str, file_id: int = None):
|
||||
"This should only be used in development/testing. For production, remove "
|
||||
"SFTP_DISABLE_HOST_KEY_VERIFICATION or set it to False and configure known_hosts."
|
||||
)
|
||||
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # nosec B507 - Configurable, warns user
|
||||
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # nosec B507 - Configurable, warns user # noqa: S507
|
||||
else:
|
||||
# Use system known_hosts for host key verification (more secure)
|
||||
ssh.load_system_host_keys()
|
||||
@@ -149,8 +149,8 @@ def upload_to_sftp(self, file_path: str, file_id: int = None):
|
||||
if "sftp" in locals():
|
||||
sftp.close()
|
||||
ssh.close()
|
||||
except Exception:
|
||||
pass
|
||||
except Exception: # noqa: S110
|
||||
pass # Ignore errors during cleanup
|
||||
|
||||
error_msg = f"Failed to upload {filename} to SFTP server: {str(e)}"
|
||||
logger.error(f"[{task_id}] {error_msg}")
|
||||
|
||||
@@ -57,7 +57,7 @@ def upload_with_rclone(self, file_path: str, destination: str):
|
||||
# Ensure the remote path exists (create folders if needed)
|
||||
mkdir_cmd = ["rclone", "mkdir", "--config", rclone_config_path, destination]
|
||||
|
||||
subprocess.run(mkdir_cmd, check=True, capture_output=True)
|
||||
subprocess.run(mkdir_cmd, check=True, capture_output=True) # noqa: S603
|
||||
|
||||
# Construct the upload command
|
||||
upload_cmd = ["rclone", "copy", "--config", rclone_config_path, file_path, destination, "--progress"]
|
||||
@@ -65,14 +65,14 @@ def upload_with_rclone(self, file_path: str, destination: str):
|
||||
log_task_progress(task_id, "rclone_upload", "in_progress", f"Executing rclone copy to {destination}")
|
||||
|
||||
# Execute the upload command
|
||||
result = subprocess.run(upload_cmd, check=True, capture_output=True, text=True)
|
||||
result = subprocess.run(upload_cmd, check=True, capture_output=True, text=True) # noqa: S603
|
||||
|
||||
# Check if upload was successful
|
||||
if result.returncode == 0:
|
||||
# Try to get a public link if possible
|
||||
try:
|
||||
link_cmd = ["rclone", "link", "--config", rclone_config_path, f"{destination}/{filename}"]
|
||||
link_result = subprocess.run(link_cmd, capture_output=True, text=True, check=False)
|
||||
link_result = subprocess.run(link_cmd, capture_output=True, text=True, check=False) # noqa: S603
|
||||
public_url = link_result.stdout.strip() if link_result.returncode == 0 else None
|
||||
except (subprocess.SubprocessError, OSError) as e:
|
||||
logger.warning(f"[{task_id}] Failed to get public link for {filename}: {str(e)}")
|
||||
@@ -135,7 +135,7 @@ def send_to_all_rclone_destinations(self, file_path: str):
|
||||
# Get list of configured destinations from rclone
|
||||
try:
|
||||
remotes_cmd = ["rclone", "listremotes", "--config", rclone_config_path]
|
||||
result = subprocess.run(remotes_cmd, check=True, capture_output=True, text=True)
|
||||
result = subprocess.run(remotes_cmd, check=True, capture_output=True, text=True) # noqa: S603
|
||||
|
||||
if result.returncode == 0:
|
||||
# Process the list of remotes
|
||||
|
||||
Reference in New Issue
Block a user