style(project): 💄 update docstrings with ruff

change ruff configurations in pyproject.toml in order to improve docstring format in project
This commit is contained in:
2026-09-15 16:33:45 +02:00
parent 20ad43d694
commit eb4ef2339e
4 changed files with 66 additions and 45 deletions
+8 -1
View File
@@ -39,4 +39,11 @@ exclude = [
[[tool.mypy.overrides]] [[tool.mypy.overrides]]
module = "exchangelib.*" module = "exchangelib.*"
ignore_missing_imports = true ignore_missing_imports = true
[tool.ruff.format]
docstring-code-format = true
[tool.ruff.lint]
select = ["D"]
ignore = ["D100"]
+19 -10
View File
@@ -27,8 +27,7 @@ logger = logging.getLogger("logwatcher")
def _is_log_file(path: Path) -> bool: def _is_log_file(path: Path) -> bool:
""" """Return True if file is a log file (.txt, .log, or extensionless).
Return True if file is a log file (.txt, .log, or extensionless).
Args: Args:
path: Path of a file path: Path of a file
@@ -36,6 +35,7 @@ def _is_log_file(path: Path) -> bool:
Returns: Returns:
True if file has .txt or .log extension or extensionless True if file has .txt or .log extension or extensionless
False otherwise. False otherwise.
""" """
if not path.is_file(): if not path.is_file():
return False return False
@@ -61,12 +61,13 @@ def _files_to_log_entries(input_files: list[Path]) -> list[LogEntry]:
def version_callback(value: bool): def version_callback(value: bool):
""" """Print the version of package.
Print the version of package.
Stop the CLI after printing version. Stop the CLI after printing version.
Args: Args:
value: True if '--version' is provided, False otherwise. value: True if '--version' is provided, False otherwise.
""" """
if value: if value:
print(f"logwatcher version: {__version__}") print(f"logwatcher version: {__version__}")
@@ -82,6 +83,15 @@ def from_mails(
), ),
] = Path("output"), ] = Path("output"),
): ):
"""Convert mail containing logs into reports for N2 technicians.
Connect to the EWS server with service account, filter most recent mail logs,
generate reports based on these mails and send mails to the technicians.
Args:
output_dir: Path to the generated reports folder
"""
logger.info("============ logwatcher mail started ============") logger.info("============ logwatcher mail started ============")
account = connect_to_mailbox() account = connect_to_mailbox()
messages = fetch_log_messages(account) messages = fetch_log_messages(account)
@@ -113,7 +123,7 @@ def from_mails(
start_date=start_date, start_date=start_date,
end_date=end_date, end_date=end_date,
) )
send_n2_report(mail_body, account, output_dir) send_n2_report(account, mail_body, output_dir / "n2.log")
move_analyzed_mails(account, messages) move_analyzed_mails(account, messages)
logger.info("============ logwatcher mail ended ============") logger.info("============ logwatcher mail ended ============")
@@ -135,8 +145,8 @@ def from_files(
typer.Option("--input-dir", help="Directories containing log files."), typer.Option("--input-dir", help="Directories containing log files."),
] = None, ] = None,
): ):
""" """Convert log files into two reports, one of n2 related logs, one without any n2 related logs.
Convert log files into two reports, one of n2 related logs, one without any n2 related logs.
These reports are sent by mail to n2 support technicians. These reports are sent by mail to n2 support technicians.
These log entries are used to generate incidents on Ivanti using its API. These log entries are used to generate incidents on Ivanti using its API.
@@ -148,6 +158,7 @@ def from_files(
Returns: Returns:
None None
""" """
logger.info("============ logwatcher file started ============") logger.info("============ logwatcher file started ============")
if input_files and input_dir: if input_files and input_dir:
@@ -204,14 +215,12 @@ def cli(
), ),
] = False, ] = False,
) -> None: ) -> None:
""" """Entrypoint for logwatcher.
Entrypoint for logwatcher.
Args: Args:
verbose: if provided, script returns info and debug logs too. Otherwise, it returns only warnings and errors. verbose: if provided, script returns info and debug logs too. Otherwise, it returns only warnings and errors.
version: if provided, script returns the version of package version: if provided, script returns the version of package
Returns:
""" """
setup_logging(verbose=verbose) setup_logging(verbose=verbose)
+19 -22
View File
@@ -23,8 +23,7 @@ LOG_IN_ATTACHMENT_PATTERN = "Le compte-rendu contient plus de 100 lignes."
def _find_log_attachment(mail: Message) -> FileAttachment | None: def _find_log_attachment(mail: Message) -> FileAttachment | None:
""" """Find the right log file attachment among other content such as images from footers.
Find the right log file attachment among other content such as images from footers.
Mail is expected to have one matching attachment whose name starts with Mail is expected to have one matching attachment whose name starts with
'CR_' and has one of these extensions: txt, log or no extensions. 'CR_' and has one of these extensions: txt, log or no extensions.
@@ -34,6 +33,7 @@ def _find_log_attachment(mail: Message) -> FileAttachment | None:
Returns: Returns:
the attachment found or None the attachment found or None
""" """
for attachment in mail.attachments: for attachment in mail.attachments:
if isinstance(attachment, FileAttachment) and not attachment.is_inline: if isinstance(attachment, FileAttachment) and not attachment.is_inline:
@@ -61,15 +61,14 @@ def _get_attachment_content(attachment: FileAttachment) -> str:
def _get_mail_body(mail: Message) -> str: def _get_mail_body(mail: Message) -> str:
""" """Find the start of the log content and retrieve everything that follows.
Find the start of the log content and retrieve everything
that follows.
Args: Args:
mail: log related mail mail: log related mail
Returns: Returns:
part of the body of the log message part of the body of the log message
""" """
marker = "Répertoire scanné" marker = "Répertoire scanné"
index = mail.text_body.find(marker) index = mail.text_body.find(marker)
@@ -80,8 +79,7 @@ def _get_mail_body(mail: Message) -> str:
def _has_attachment_marker(body: str) -> bool: def _has_attachment_marker(body: str) -> bool:
""" """Return True if the email says the report is in the attachment.
Return True if the email says the report is in the attachment.
Args: Args:
body: the email body text. body: the email body text.
@@ -89,24 +87,24 @@ def _has_attachment_marker(body: str) -> bool:
Returns: Returns:
True if the email says the report is in the attachment. True if the email says the report is in the attachment.
False otherwise. False otherwise.
""" """
return LOG_IN_ATTACHMENT_PATTERN in body return LOG_IN_ATTACHMENT_PATTERN in body
def connect_to_mailbox() -> Account: def connect_to_mailbox() -> Account:
""" """Connect to mailbox and verify connection with a server roundtrip.
Connect to mailbox and verify connection with a server roundtrip.
Returns: Returns:
the account associated with the mailbox containing log files and log mails the account associated with the mailbox containing log files and log mails
""" """
try: try:
email = os.environ["EMAIL"] email = os.environ["EMAIL"]
password = os.environ["PASSWORD"] password = os.environ["PASSWORD"]
endpoint = os.environ["EWS_URL"] endpoint = os.environ["EWS_URL"]
except KeyError as error: except KeyError as error:
error_message = f"'{error.args}' was not found in environment variables." logger.error(f"'{error.args[0]}' was not found in environment variables.")
logger.error(error_message)
raise raise
credentials = Credentials(username=email, password=password) credentials = Credentials(username=email, password=password)
@@ -130,39 +128,38 @@ def connect_to_mailbox() -> Account:
def fetch_log_messages(account: Account) -> list[Message]: def fetch_log_messages(account: Account) -> list[Message]:
""" """Retrieve log messages from the log folder.
Retrieve log messages from the log folder.
Args: Args:
account: account object that owns the mailbox of log related mails. account: account object that owns the mailbox of log related mails.
Returns: Returns:
queryset of log-related mails queryset of log-related mails
""" """
logs_folder = account.msg_folder_root / "Logs" logs_folder = account.msg_folder_root / "Logs"
return list(logs_folder.all()) return list(logs_folder.all())
def extract_log_lines(content: str) -> list[str]: def extract_log_lines(content: str) -> list[str]:
""" """Split raw log content into clean, non-empty log lines and return a list of clean log line.
Split raw log content into clean, non-empty log lines
and return a list of clean log line.
Args: Args:
content: string of raw logs content: string of raw logs
Returns: Returns:
list of logs list of logs
""" """
return [line.strip("\r\n") for line in content.splitlines() if line.strip()] return [line.strip("\r\n") for line in content.splitlines() if line.strip()]
def extract_logs_from_mails(messages: list[Message]) -> list[str]: def extract_logs_from_mails(messages: list[Message]) -> list[str]:
""" """Retrieve emails that are N2 log-related, clean them and return a list of logs in string format.
Retrieves emails that are N2 log-related,
clean them and return a list of logs in string format. Returns:
list of logs in string format
Returns: list of logs in string format
""" """
logs = [] logs = []
@@ -182,8 +179,7 @@ def extract_logs_from_mails(messages: list[Message]) -> list[str]:
def move_analyzed_mails(account: Account, messages: list[Message]) -> None: def move_analyzed_mails(account: Account, messages: list[Message]) -> None:
""" """Move processed mails from the Logs folder to the Analyzed subfolder.
Move processed mails from the Logs folder to the Analyzed subfolder.
Args: Args:
account: The mailbox account owning the Logs folder. account: The mailbox account owning the Logs folder.
@@ -192,6 +188,7 @@ def move_analyzed_mails(account: Account, messages: list[Message]) -> None:
Note: Note:
Partial failure is possible: if a move fails mid-batch, some mails Partial failure is possible: if a move fails mid-batch, some mails
may remain in Logs and be retried on the next run (at-least-once). may remain in Logs and be retried on the next run (at-least-once).
""" """
if not messages: if not messages:
logger.info("No mails to move.") logger.info("No mails to move.")
+20 -12
View File
@@ -6,33 +6,41 @@ from exchangelib import Account, FileAttachment, Message
from logwatcher.utils import get_or_create_folder from logwatcher.utils import get_or_create_folder
LOG_FOLDER = "Logs"
SENT_FOLDER = "Sent"
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def send_n2_report(summary: str, account: Account, output_path: Path): def send_n2_report(account: Account, summary: str, n2_log_file: Path) -> None:
to = os.environ["N2_REPORT_RECIPIENTS"] """Send the N2 report by e-mail through EWS.
subject = os.environ["N2_REPORT_SUBJECT"]
Args:
# generate file attachments account: the EWS account used to send email report from
n2_log_file = output_path / "n2.log" summary: text body of the email report
n2_log_file: path to the n2 report file
"""
try:
to = os.environ["N2_REPORT_RECIPIENTS"].split(",")
subject = os.environ["N2_REPORT_SUBJECT"]
except KeyError as error:
logger.error(f"{error.args[0]} was not found in environment variable.")
raise
# build file attachment
logger.info(f"Create attachment based on '{n2_log_file}' file.") logger.info(f"Create attachment based on '{n2_log_file}' file.")
n2_log_attachment = FileAttachment(name=n2_log_file.name, content=n2_log_file.read_bytes(), is_inline=False) n2_log_attachment = FileAttachment(name=n2_log_file.name, content=n2_log_file.read_bytes(), is_inline=False)
logger.info(f"Attachment '{n2_log_attachment.name}' was successfully created.") logger.info(f"Attachment '{n2_log_attachment.name}' was successfully created.")
# get sent folder # get sent folder
sent_folder = get_or_create_folder(account, SENT_FOLDER) sent_folder = get_or_create_folder(account, "Sent")
# create mail # create mail
logger.info(f"Create mail with account='{account.fullname}', recipients='{to}', subject='{subject}'.") logger.info(f"Create mail with account='{account.fullname}', recipients='{to}', subject='{subject}'.")
mail = Message(account=account, to_recipients=to, subject=subject, body=summary) mail = Message(account=account, to_recipients=to, subject=subject, body=summary)
mail.attach(n2_log_attachment) mail.attach(n2_log_attachment)
logger.info(f"Mail '{mail.subject}' was successfully created with attachment '{mail.attachments[0].name}'.") logger.info(f"Mail '{mail.subject}' was successfully created with attachment '{n2_log_attachment.name}'.")
# send and keep a copy to the Logs/Sent folder # send and keep a copy to the Logs/Sent folder
logger.info(f"Attempt to send mail '{mail.subject}'...") logger.info(f"Attempt to send mail '{mail.subject}'...")
mail.send(copy_to_folder=sent_folder) # TODO check if message was successfully sent mail.send(copy_to_folder=sent_folder)
logger.info(f"Mail '{subject}' was successfully sent to '{to}'.") logger.info(f"Mail '{subject}' was successfully sent to '{to}'.")
logger.info(f"Mail is saved in '{sent_folder}' folder.") logger.info(f"Mail is saved in '{sent_folder}' folder.")