Spaces:
Sleeping
Sleeping
File size: 3,513 Bytes
c55abce | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 | """Resolve a sender's email domain to the Dropbox client-folder name.
Used for routing Purchase Order and Invoice attachments under a per-client
subfolder. Unknown domains route to `_unmatched` and are surfaced in the
daily brief so ops can add a mapping without a redeploy.
The mapping table is seeded via `scripts/seed_client_domains.py` and can
be edited at runtime by inserting rows directly into the `client_domains`
table (no server restart needed — lookups happen per request).
"""
from __future__ import annotations
import logging
import re
from typing import Optional
import aiosqlite
logger = logging.getLogger(__name__)
# Conservative domain extraction. We accept lowercase letters, digits, dot,
# and hyphen — the standard DNS character set. Anything else gets rejected
# so we don't pollute the table with junk like "foo@bar baz.com".
_DOMAIN_RE = re.compile(r"^[a-z0-9.\-]+$")
def _extract_domain(sender_email: str) -> Optional[str]:
"""Return the lowercased, validated domain portion of an email, or None.
Strips angle brackets and surrounding whitespace; accepts "<a@b.com>",
" a@b.com ", "FOO@BAR.COM". Rejects malformed addresses outright.
"""
if not sender_email:
return None
s = sender_email.strip().strip("<>").lower()
if "@" not in s:
return None
_local, _, domain = s.rpartition("@")
domain = domain.strip()
if not domain or not _DOMAIN_RE.match(domain):
return None
return domain
async def resolve_client_name(
conn: aiosqlite.Connection,
sender_email: str,
) -> Optional[str]:
"""Look up the client-folder name for a sender's email domain.
Returns the client name string if a mapping exists, otherwise None.
Updates `last_seen_at` on the matched row so ops can see which mappings
are actively being used.
The conn is expected to be opened with `aiosqlite.Row` as row factory
(the standard `get_db` context manager in app.database does this).
"""
domain = _extract_domain(sender_email)
if domain is None:
logger.debug("resolve_client_name: rejected sender %r", sender_email)
return None
cursor = await conn.execute(
"SELECT client_name FROM client_domains WHERE domain = ?",
(domain,),
)
row = await cursor.fetchone()
if row is None:
return None
# Touch last_seen_at so ops can see mapping activity.
await conn.execute(
"UPDATE client_domains SET last_seen_at = datetime('now') WHERE domain = ?",
(domain,),
)
await conn.commit()
return row["client_name"]
async def add_or_update_client_domain(
conn: aiosqlite.Connection,
domain: str,
client_name: str,
notes: Optional[str] = None,
) -> None:
"""Upsert a domain→client mapping.
Used by the seed script and by any future admin tool. `domain` is
normalized lowercase; `client_name` is stored as-is (so case can be
preserved in the Dropbox folder name, e.g. "Westlake").
"""
domain_norm = (domain or "").strip().lower()
if not _DOMAIN_RE.match(domain_norm):
raise ValueError(f"Invalid domain: {domain!r}")
await conn.execute(
"""
INSERT INTO client_domains (domain, client_name, notes)
VALUES (?, ?, ?)
ON CONFLICT(domain) DO UPDATE SET
client_name = excluded.client_name,
notes = excluded.notes
""",
(domain_norm, client_name.strip(), notes),
)
await conn.commit()
|