Nicolas Cantu
09c8f87c90
**Motivations:** - Replace per-path messages with canonical doc reference from LeCoffre imap_common. **Evolutions:** - mail_common re-exports IMAP_SMTP_CONFIG_DOC_REF; mail-*.py and tickets-fetch-inbox use it for IMAP/SMTP/Gitea token errors. **Page affectées:** - git-issues/mail_common.py, mail-list-unread.py, mail-get-thread.py, mail-mark-read.py, mail-send-reply.py, mail-to-issue.py, mail-create-issue-from-email.py, tickets-fetch-inbox.py
117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
# Shared config and helpers for git-issues mail scripts (Gitea + re-exported IMAP/SMTP from LeCoffre).
|
|
# IMAP/SMTP implementation: lecoffre_ng_test/automation/imap-bridge/imap_common.py
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
# Must match automation/imap-bridge/imap_common.IMAP_SMTP_CONFIG_DOC_REF (used before imap_common is importable).
|
|
_BOOTSTRAP_IMAP_DOC_REF = "voir automation/imap-bridge/README.md"
|
|
|
|
|
|
def _imap_bridge_python_dir() -> Path:
|
|
if os.environ.get("LECOFFRE_REPO_ROOT"):
|
|
p = Path(os.environ["LECOFFRE_REPO_ROOT"]).expanduser().resolve() / "automation" / "imap-bridge"
|
|
if (p / "imap_common.py").is_file():
|
|
return p
|
|
here = Path(__file__).resolve().parent
|
|
ia_dev_root = here.parent
|
|
sibling = ia_dev_root.parent / "lecoffre_ng_test" / "automation" / "imap-bridge"
|
|
if (sibling / "imap_common.py").is_file():
|
|
return sibling
|
|
raise ImportError(
|
|
"imap_common not found: set LECOFFRE_REPO_ROOT to the LeCoffre monorepo root "
|
|
"or clone lecoffre_ng_test beside ia_dev (../lecoffre_ng_test/automation/imap-bridge/imap_common.py). "
|
|
+ _BOOTSTRAP_IMAP_DOC_REF
|
|
)
|
|
|
|
|
|
_pkg = _imap_bridge_python_dir()
|
|
if str(_pkg) not in sys.path:
|
|
sys.path.insert(0, str(_pkg))
|
|
|
|
from imap_common import (
|
|
IMAP_SMTP_CONFIG_DOC_REF,
|
|
MAIL_SINCE_DATE_DEFAULT,
|
|
imap_search_criterion_all,
|
|
imap_search_criterion_unseen,
|
|
imap_since_date,
|
|
imap_ssl_context,
|
|
load_imap_config,
|
|
load_smtp_config,
|
|
)
|
|
|
|
|
|
def repo_root() -> Path:
|
|
env_root = os.environ.get("REPO_ROOT")
|
|
if env_root:
|
|
return Path(env_root).resolve()
|
|
issues_dir = os.environ.get("GIT_ISSUES_DIR")
|
|
if issues_dir:
|
|
return Path(issues_dir).resolve().parent
|
|
return Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def _lecoffre_automation_token_path() -> Path | None:
|
|
lr = os.environ.get("LECOFFRE_REPO_ROOT", "").strip()
|
|
if lr:
|
|
p = Path(lr).expanduser().resolve() / ".secrets" / "automation" / "git-issues" / "token"
|
|
return p if p.is_file() else None
|
|
sib = repo_root().parent / "lecoffre_ng_test" / ".secrets" / "automation" / "git-issues" / "token"
|
|
return sib if sib.is_file() else None
|
|
|
|
|
|
def load_gitea_config() -> dict[str, str]:
|
|
root = repo_root()
|
|
token = os.environ.get("GITEA_TOKEN")
|
|
if not token:
|
|
lt = _lecoffre_automation_token_path()
|
|
if lt is not None:
|
|
token = lt.read_text(encoding="utf-8").strip()
|
|
if not token:
|
|
for sub in ("git-issues", "gitea-issues"):
|
|
token_path = root / ".secrets" / sub / "token"
|
|
if token_path.is_file():
|
|
token = token_path.read_text(encoding="utf-8").strip()
|
|
break
|
|
return {
|
|
"api_url": os.environ.get("GITEA_API_URL", "https://git.4nkweb.com/api/v1").rstrip("/"),
|
|
"owner": os.environ.get("GITEA_REPO_OWNER", "4nk"),
|
|
"repo": os.environ.get("GITEA_REPO_NAME", "lecoffre_ng"),
|
|
"token": token or "",
|
|
}
|
|
|
|
|
|
def sanitize_title(raw: str, max_len: int = 200) -> str:
|
|
one_line = re.sub(r"\s+", " ", raw).strip()
|
|
return one_line[:max_len] if one_line else "(no subject)"
|
|
|
|
|
|
def create_gitea_issue(title: str, body: str) -> dict | None:
|
|
gitea = load_gitea_config()
|
|
if not gitea["token"]:
|
|
return None
|
|
url = f"{gitea['api_url']}/repos/{gitea['owner']}/{gitea['repo']}/issues"
|
|
payload = json.dumps({"title": title, "body": body}).encode("utf-8")
|
|
req = Request(
|
|
url,
|
|
data=payload,
|
|
method="POST",
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"token {gitea['token']}",
|
|
},
|
|
)
|
|
try:
|
|
with urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
except (HTTPError, URLError):
|
|
return None
|