Compare commits

...

5 Commits

Author SHA1 Message Date
Clemens Schwaighofer
0e6331fa6a v0.33.0: datetime parsing update 2025-11-06 13:26:07 +09:00
Clemens Schwaighofer
c98c5df63c Update datetime parse helper
Allow non T in isotime format, add non T normal datetime parsing
2025-11-06 13:24:27 +09:00
Clemens Schwaighofer
0981c74da9 v0.32.0: add email sending 2025-10-27 11:22:11 +09:00
Clemens Schwaighofer
31518799f6 README update 2025-10-27 11:20:46 +09:00
Clemens Schwaighofer
e8b4b9b48e Add send email class 2025-10-27 11:19:38 +09:00
9 changed files with 1512 additions and 10 deletions

View File

@@ -10,6 +10,7 @@ This is a pip package that can be installed into any project and covers the foll
- logging update with exception logs
- requests wrapper for easier auth pass on access
- dict fingerprinting
- sending email
- jmespath search
- json helpers for conten replace and output
- dump outputs for data for debugging
@@ -26,6 +27,7 @@ This is a pip package that can be installed into any project and covers the foll
- debug_handling: various debug helpers like data dumper, timer, utilization, etc
- db_handling: SQLite interface class
- encyption_handling: symmetric encryption
- email_handling: simple email sending
- file_handling: crc handling for file content and file names, progress bar
- json_handling: jmespath support and json date support, replace content in dict with json paths
- iterator_handling: list and dictionary handling support (search, fingerprinting, etc)
@@ -50,7 +52,7 @@ Have the following setup in `project.toml`
```toml
[[tool.uv.index]]
name = "egra-gitea"
name = "opj-pypi"
url = "https://git.egplusww.jp/api/packages/PyPI/pypi/simple/"
publish-url = "https://git.egplusww.jp/api/packages/PyPI/pypi"
explicit = true
@@ -58,15 +60,15 @@ explicit = true
```sh
uv build
uv publish --index egra-gitea --token <gitea token>
uv publish --index opj-pypi --token <gitea token>
```
## Test package
## Use package
We must set the full index URL here because we run with "--no-project"
```sh
uv run --with corelibs --index egra-gitea=https://git.egplusww.jp/api/packages/PyPI/pypi/simple/ --no-project -- python -c "import corelibs"
uv run --with corelibs --index opj-pypi=https://git.egplusww.jp/api/packages/PyPI/pypi/simple/ --no-project -- python -c "import corelibs"
```
### Python tests
@@ -99,7 +101,7 @@ uv run test-run/<script>
This will also add the index entry
```sh
uv add corelibs --index egra-gitea=https://git.egplusww.jp/api/packages/PyPI/pypi/simple/
uv add corelibs --index opj-pypi=https://git.egplusww.jp/api/packages/PyPI/pypi/simple/
```
## Python venv setup

View File

@@ -1,7 +1,7 @@
# MARK: Project info
[project]
name = "corelibs"
version = "0.31.1"
version = "0.33.0"
description = "Collection of utils for Python scripts"
readme = "README.md"
requires-python = ">=3.13"
@@ -17,7 +17,7 @@ dependencies = [
# MARK: build target
[[tool.uv.index]]
name = "egra-gitea"
name = "opj-pypi"
url = "https://git.egplusww.jp/api/packages/PyPI/pypi/simple/"
publish-url = "https://git.egplusww.jp/api/packages/PyPI/pypi"
explicit = true
@@ -63,12 +63,13 @@ ignore = [
[tool.pylint.MASTER]
# this is for the tests/etc folders
init-hook='import sys; sys.path.append("src/")'
# MARK: Testing
[tool.pytest.ini_options]
testpaths = [
"tests",
]
[tool.coverage.run]
omit = [
"*/tests/*",

View File

@@ -159,10 +159,14 @@ def parse_flexible_date(
# Try different parsing methods
parsers: list[Callable[[str], datetime]] = [
# ISO 8601 format
# ISO 8601 format, also with missing "T"
lambda x: datetime.fromisoformat(x), # pylint: disable=W0108
lambda x: datetime.fromisoformat(x.replace(' ', 'T')), # pylint: disable=W0108
# Simple date format
lambda x: datetime.strptime(x, "%Y-%m-%d"),
# datetime without T
lambda x: datetime.strptime(x, "%Y-%m-%d %H:%M:%S"),
lambda x: datetime.strptime(x, "%Y-%m-%d %H:%M:%S.%f"),
# Alternative ISO formats (fallback)
lambda x: datetime.strptime(x, "%Y-%m-%dT%H:%M:%S"),
lambda x: datetime.strptime(x, "%Y-%m-%dT%H:%M:%S.%f"),

View File

View File

@@ -0,0 +1,199 @@
"""
Send email wrapper
"""
import smtplib
from email.message import EmailMessage
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from corelibs.logging_handling.log import Logger
class SendEmail:
"""
send emails based on a template to a list of receivers
"""
def __init__(
self,
log: "Logger",
settings: dict[str, Any],
template: dict[str, str],
from_email: str,
combined_send: bool = True,
receivers: list[str] | None = None,
data: list[dict[str, str]] | None = None,
):
"""
init send email class
Args:
template (dict): Dictionary with body and subject
from_email (str): from email as "Name" <email>
combined_send (bool): True for sending as one set for all receivers
receivers (list): list of emails to send to
data (dict): data to replace in template
args (Namespace): _description_
"""
self.log = log
self.settings = settings
# internal settings
self.template = template
self.from_email = from_email
self.combined_send = combined_send
self.receivers = receivers
self.data = data
def send_email(
self,
data: list[dict[str, str]] | None,
receivers: list[str] | None,
template: dict[str, str] | None = None,
from_email: str | None = None,
combined_send: bool | None = None,
test_only: bool | None = None
):
"""
build email and send
Arguments:
data {list[dict[str, str]] | None} -- _description_
receivers {list[str] | None} -- _description_
combined_send {bool | None} -- _description_
Keyword Arguments:
template {dict[str, str] | None} -- _description_ (default: {None})
from_email {str | None} -- _description_ (default: {None})
Raises:
ValueError: _description_
ValueError: _description_
"""
if data is None and self.data is not None:
data = self.data
if data is None:
raise ValueError("No replace data set, cannot send email")
if receivers is None and self.receivers is not None:
receivers = self.receivers
if receivers is None:
raise ValueError("No receivers list set, cannot send email")
if combined_send is None:
combined_send = self.combined_send
if test_only is not None:
self.settings['test'] = test_only
if template is None:
template = self.template
if from_email is None:
from_email = self.from_email
if not template['subject'] or not template['body']:
raise ValueError("Both Subject and Body must be set")
self.log.debug(
"[EMAIL]:\n"
f"Subject: {template['subject']}\n"
f"Body: {template['body']}\n"
f"From: {from_email}\n"
f"Combined send: {combined_send}\n"
f"Receivers: {receivers}\n"
f"Replace data: {data}"
)
# send email
self.send_email_list(
self.prepare_email_content(
from_email, template, data
),
receivers,
combined_send,
test_only
)
def prepare_email_content(
self,
from_email: str,
template: dict[str, str],
data: list[dict[str, str]],
) -> list[EmailMessage]:
"""
prepare email for sending
Args:
template (dict): template data for this email
data (dict): data to replace in email
Returns:
list: Email Message Objects as list
"""
_subject = ""
_body = ""
msg: list[EmailMessage] = []
for replace in data:
_subject = template["subject"]
_body = template["body"]
for key, value in replace.items():
_subject = _subject.replace(f"{{{{{key}}}}}", value)
_body = _body.replace(f"{{{{{key}}}}}", value)
# create a simple email and add subhect, from email
msg_email = EmailMessage()
# msg.set_content(_body, charset='utf-8', cte='quoted-printable')
msg_email.set_content(_body, charset="utf-8")
msg_email["Subject"] = _subject
msg_email["From"] = from_email
# push to array for sening
msg.append(msg_email)
return msg
def send_email_list(
self,
email: list[EmailMessage], receivers: list[str],
combined_send: bool | None = None,
test_only: bool | None = None
):
"""
send email to receivers list
Args:
email (list): Email Message object with set obdy, subject, from as list
receivers (array): email receivers list as array
combined_send (bool): True for sending as one set for all receivers
"""
if test_only is not None:
self.settings['test'] = test_only
# localhost (postfix does the rest)
smtp = None
smtp_host = self.settings.get('smtp_host', "localhost")
try:
smtp = smtplib.SMTP(smtp_host)
except ConnectionRefusedError as e:
self.log.error("Could not open SMTP connection to: %s, %s", smtp_host, e)
# loop over messages and then over recievers
for msg in email:
if combined_send is True:
msg["To"] = ", ".join(receivers)
if not self.settings.get('test'):
if smtp is not None:
smtp.send_message(msg, msg["From"], receivers)
else:
self.log.info(f"[EMAIL] Test, not sending email\n{msg}")
else:
for receiver in receivers:
# send to
self.log.debug(f"===> Send to: {receiver}")
if "To" in msg:
msg.replace_header("To", receiver)
else:
msg["To"] = receiver
if not self.settings.get('test'):
if smtp is not None:
smtp.send_message(msg)
else:
self.log.info(f"[EMAIL] Test, not sending email\n{msg}")
# close smtp
if smtp is not None:
smtp.quit()
# __END__

View File

View File

@@ -275,6 +275,53 @@ class TestParseFlexibleDate:
assert isinstance(result, datetime)
assert result.tzinfo is not None
def test_parse_flexible_date_missing_t_with_timezone_shift(self):
"""Test parse_flexible_date with timezone shift"""
result = parse_flexible_date('2023-12-25 15:30:45+00:00', timezone_tz='Asia/Tokyo', shift_time_zone=True)
assert isinstance(result, datetime)
assert result.tzinfo is not None
def test_parse_flexible_date_space_separated_datetime(self):
"""Test parse_flexible_date with space-separated datetime format"""
result = parse_flexible_date('2023-12-25 15:30:45')
assert isinstance(result, datetime)
assert result.year == 2023
assert result.month == 12
assert result.day == 25
assert result.hour == 15
assert result.minute == 30
assert result.second == 45
def test_parse_flexible_date_space_separated_with_microseconds(self):
"""Test parse_flexible_date with space-separated datetime and microseconds"""
result = parse_flexible_date('2023-12-25 15:30:45.123456')
assert isinstance(result, datetime)
assert result.year == 2023
assert result.month == 12
assert result.day == 25
assert result.hour == 15
assert result.minute == 30
assert result.second == 45
assert result.microsecond == 123456
def test_parse_flexible_date_t_separated_datetime(self):
"""Test parse_flexible_date with T-separated datetime (alternative ISO format)"""
result = parse_flexible_date('2023-12-25T15:30:45')
assert isinstance(result, datetime)
assert result.year == 2023
assert result.month == 12
assert result.day == 25
assert result.hour == 15
assert result.minute == 30
assert result.second == 45
def test_parse_flexible_date_t_separated_with_microseconds(self):
"""Test parse_flexible_date with T-separated datetime and microseconds"""
result = parse_flexible_date('2023-12-25T15:30:45.123456')
assert isinstance(result, datetime)
assert result.year == 2023
assert result.microsecond == 123456
def test_parse_flexible_date_invalid_format(self):
"""Test parse_flexible_date with invalid format returns None"""
result = parse_flexible_date('invalid-date')

File diff suppressed because it is too large Load Diff

2
uv.lock generated
View File

@@ -108,7 +108,7 @@ wheels = [
[[package]]
name = "corelibs"
version = "0.31.1"
version = "0.33.0"
source = { editable = "." }
dependencies = [
{ name = "cryptography" },