wr-common-lib

Shared enums, email helpers, geospatial coordinates, formatting utilities, and optional PostgreSQL write helpers.

Requires Python 3.12+.

Install

pip install wr-common-lib
pip install "wr-common-lib[resend]"   # ResendProvider (httpx)
pip install "wr-common-lib[db]"       # EmailDbOper, AisDbOper (async-db-tools)
pip install "wr-common-lib[db,resend]"

Package layout

src/wr_common_lib/
├── __init__.py          # __version__
├── email/
│   ├── __init__.py
│   ├── constants.py     # MailFlow, MailStatus, VoyageStatus, PassVia, CustomerRole
│   ├── encoding.py      # file_to_base64, file_to_attachment
│   ├── provider.py      # ResendProvider, SendResult, ReceivedEmail
│   └── db_oper.py       # EmailDbOper, get_email_content_hash, …
├── geo/
│   ├── coordinates.py   # DMS parse/format, longitude continuity
│   └── geo_point.py     # GeoPoint
├── ais/
│   ├── models.py        # AisPoint
│   ├── thin.py          # thin_ais_points（内存回退）
│   └── postgres.py      # AisDbOper（vessel_ais 库内抽稀）
└── utils/
    ├── algorithm.py     # find_nearest_idx
    ├── convert.py       # as_uuid, round_precision, json/dict, …
    ├── file.py          # is_valid_size_file, is_recently_modified_file, is_valid_grib2_file
    ├── formatters.py    # format_number, format_timestamp, yyyymmddhh_to_datetime, …
    ├── http.py          # fetch_text（需 httpx）
    └── time.py          # ensure_utc, to_unix_seconds, …

Content hash

from wr_common_lib.email import get_email_content_hash

content_hash = get_email_content_hash(
    to="a@example.com,b@example.com",
    subject="...",
    content="...",
    cc="",
    attachments=[{"filename": "report.pdf"}],
)

Normalizes to / cc (comma-separated), strips subject/content, and hashes attachment filenames only (not file bytes).

Inbound parsing helpers

from wr_common_lib.email import extract_ship_candidates, extract_imo_from_content

extract_ship_candidates("MV OCEAN STAR - noon report")
extract_imo_from_content("IMO 9682930")   # "9682930" or None

Attachment encoding

Converts local files to Base64 for Resend attachments[].content.

from wr_common_lib.email import file_to_base64, file_to_attachment

b64 = file_to_base64("/path/to/report.pdf")

att = file_to_attachment("/path/to/report.pdf")
# {"filename": "report.pdf", "content": "<base64>", "content_type": "application/pdf"}

ResendProvider

Requires httpx ([resend] extra). Uses the Resend API.

from wr_common_lib.email import EmailProvider, ResendProvider, file_to_attachment

provider = ResendProvider(
    api_key="re_...",
    default_from="noreply@example.com",
    default_from_name="My App",
)

# Send (attachments from file_to_attachment)
result = await provider.send(
    to="user@example.com",
    cc="",
    subject="Hello",
    content="Plain text body",
    attachments=[file_to_attachment("/path/to/file.pdf")],
    mail_from=None,           # optional override
    email_id=email_id,        # optional, sets X-Email-ID header
)
# result.ok, result.message_id

# Inbound (Receiving API)
mail = await provider.received(message_id)
# mail.ok, mail.subject, mail.text, mail.html, mail.from_addr, mail.to, mail.cc, ...
body = mail.content   # text, else html

EmailDbOper

Requires the [db] extra. Pass an async-db-tools PostgresPool (or compatible pool with fetchval / execute).

from wr_common_lib.email import EmailDbOper, MailFlow, MailStatus, get_email_content_hash

db_oper = EmailDbOper(pool)

Write operations only; reads should use db_oper._db directly (e.g. fetchrow, fetchval).

insert_email

email_id, created = await db_oper.insert_email(
    task,
    MailFlow.OUTBOUND,
    MailStatus.PENDING,
)
# created is True when a new row was inserted, False when content_hash already existed

Task fields (inbound and outbound use the same shape):

On duplicate content_hash, the insert is skipped (ON CONFLICT DO NOTHING); the existing row id is returned and created is False.

update_status / update_parsed_data

await db_oper.update_status(email_id, MailStatus.SENT)
await db_oper.update_parsed_data(email_id, {"parsed": "..."})

Geo

Decimal degrees, DMS parsing/formatting, and anti-meridian track continuity. No extra dependencies.

from wr_common_lib.geo import (
    GeoPoint,
    parse_longitude,
    parse_latitude,
    longitude_to_dms,
    make_longitude_continuous,
)

point = GeoPoint(lng=116.403874, lat=39.914883)
point = GeoPoint.from_string("116°24′14E", "39°54′53N")

point.lng_dms.formatted   # e.g. 116°24′14E
point.to_dict()           # {"lng": ..., "lat": ...}

track = make_longitude_continuous([{"lon": 170, "lat": 30}, {"lon": -170, "lat": 31}])

AIS

读取 vessel_ais 轨迹（可选 PostgreSQL 库内时间桶抽稀）。需 wr-common-lib[db]。

from datetime import UTC, datetime

from wr_common_lib.ais import AisDbOper, AisPoint, ais_point_from_row

# 领域模型（与 vessel_ais 行对应）
point = AisPoint(
    imo=9123456,
    tz=datetime(2025, 1, 1, 12, 0, tzinfo=UTC),
    lon=120.0,
    lat=30.0,
    updated_at=int(datetime(2025, 1, 2, 8, 15, 0, tzinfo=UTC).timestamp()),
)
point.position_time   # Unix 秒，由 tz 推导
point.updated_at      # Unix 秒，由 DB updated_at 推导

# 查询行 → AisPoint（updated_at timestamptz → Unix 秒）
point = ais_point_from_row(row_dict)

# PostgreSQL
oper = AisDbOper(db)
track = await oper.fetch_track(imo, start_tz, end_tz, thin_minutes=10)

Utils

from wr_common_lib.utils import (
    coerce_to_datetime,
    ensure_utc,
    find_nearest_idx,
    format_number,
    format_timestamp,
    format_timezone_offset,
    iso_to_datetime,
    norm_meteo_source,
    round_precision,
    to_unix_seconds,
    unix_interval_hours,
)

format_number(3.1415, integer_digits=1, decimal_digits=2)  # "3.14"
format_timezone_offset(8.5)                                  # "+08:30"
format_timestamp(1710000000)                                 # ISO UTC string
ensure_utc(datetime.now())                                   # UTC aware

find_nearest_idx([10, 20, 30], 20)   # [1] 精确匹配
find_nearest_idx([10, 20, 30], 15)   # [0, 1] 左右邻居
find_nearest_idx([10, 20, 30], 5)    # [0] 小于最小值
find_nearest_idx([10, 20, 30], 35)   # [2] 大于最大值

Dependencies

Your application owns API keys, database URL, and pool lifecycle.

Publish

pip install -e ".[db,resend]"

# 1. Update CHANGELOG.md for the new version
# 2. Bump version and publish
uv version 0.2.1
export UV_PUBLISH_TOKEN=pypi-...
uv build && uv publish

git tag v0.2.1 && git push --tags

PyPI does not allow re-uploading the same version.

License

MIT — see LICENSE.