freshrss_dead_link_scanner.py

#!/usr/bin/env python3
"""
FreshRSS Dead Link Scanner
===========================
Scans a FreshRSS SQLite database for dead article URLs.

A URL is considered "dead" when:
  - HTTP 4xx client errors (404 Not Found, 410 Gone, etc.)
  - HTTP 5xx server errors (500 Internal Server Error, etc.)
  - HTTP 3xx redirect that lands on a known error page (e.g. /404/index.html)
  - Connection failures, DNS resolution failures, or timeouts

Features:
  - Filter by feed name/id and date range
  - Configurable request delay to avoid rate-limiting (429)
  - Optional: auto-tag dead URLs with a user label (我的标签) to protect them from cleanup
  - Output: terminal table (default), SMTP email notification

Prerequisites:
  pip install requests

Example usage:
  # Scan all feeds for the last 7 days, 3s delay
  python freshrss_dead_link_scanner.py ~/freshrss.db --days 7 --delay 3

  # Scan specific feeds, tag dead URLs
  python freshrss_dead_link_scanner.py freshrss.db --feed "竹新社" --start 2026-06-01 --tag "已删除"

  # With SMTP_CONFIG filled in the file, just run (no passwords on CLI):
  python freshrss_dead_link_scanner.py freshrss.db --days 30

  # Or override via CLI (--smtp-pass on CLI is discouraged):
  python freshrss_dead_link_scanner.py freshrss.db --days 30 \
      --smtp-host smtp.example.com --smtp-port 465
"""

import argparse
import contextlib
import os
import signal
import smtplib
import socket as _socket
import sqlite3
import sys
import time
import re
import textwrap
from datetime import datetime, timedelta
from email.message import EmailMessage
from typing import Optional

import requests

# ── Constants ────────────────────────────────────────────────────────────────
USER_AGENT = "FreshRSS/1.27.1 (Linux; https://freshrss.org)"

# ── SMTP defaults (edit these to avoid exposing credentials on the CLI) ──────
# CLI arguments (--smtp-*) override these values at runtime.
# Set smtp_host to an empty string ("") to disable email unless --smtp-host is
# passed explicitly.
SMTP_CONFIG = {
    "host": "",           # e.g. "smtp.gmail.com"
    "port": 587,           # 587 = STARTTLS, 465 = implicit SSL, 25 = plain
    "user": "",            # e.g. "you@gmail.com"
    "pass": "",            # App password (Gmail) or account password
    "mail_from": "",       # e.g. "you@gmail.com"
    "mail_to": "",         # comma-separated, e.g. "alice@x.com,bob@x.com"
}

# Patterns in redirect Location headers that indicate a dead / error page
DEAD_LOCATION_PATTERNS = [
    "/404", "/error", "/not-found", "/notfound",
    "404.html", "404/index", "error.html",
    "page-not-found", "not_found",
]

# HTTP status codes that unambiguously indicate a dead link
DEAD_STATUS_CODES = set(range(400, 600))  # 4xx + 5xx

# 3xx codes where we inspect the Location header
REDIRECT_CODES = {301, 302, 303, 307, 308}

# Transient network errors we retry once
RETRYABLE_EXCEPTIONS = (
    requests.Timeout,
    requests.ConnectionError,
)


# ── Dead-link detection ─────────────────────────────────────────────────────

def is_dead(response: requests.Response) -> tuple[bool, str]:
    """
    Determine if a HEAD response indicates a dead link.

    Returns (is_dead: bool, reason: str).
    """
    status = response.status_code

    # 4xx / 5xx → dead
    if status in DEAD_STATUS_CODES:
        # Preserve the real URL after redirects
        final_url = response.url
        return True, f"HTTP {status}"

    # 3xx redirect → check if it points to a known error page
    if status in REDIRECT_CODES:
        location = response.headers.get("Location", "")
        if not location:
            return False, f"HTTP {status} (no Location)"
        loc_lower = location.lower()
        for pat in DEAD_LOCATION_PATTERNS:
            if pat in loc_lower:
                return True, f"HTTP {status} → {location}"
        return False, f"HTTP {status} → {location}"

    # 2xx, 1xx, etc. → alive
    return False, f"HTTP {status}"


def check_url(
    url: str,
    timeout: float,
    max_redirects: int,
    retry: bool = True,
) -> tuple[bool, str]:
    """
    Perform a HEAD request and return (is_dead: bool, reason: str).

    Retries once on transient network errors.
    """
    headers = {"User-Agent": USER_AGENT}
    attempt = 0

    while True:
        attempt += 1
        try:
            resp = requests.head(
                url,
                headers=headers,
                timeout=timeout,
                allow_redirects=True,
                # Limit redirects to avoid loops; requests follows up to max_redirects
            )
            # requests follows redirects by default. The final status is what we get.
            # But we also want to check intermediate 3xx for error-page detection.
            # requests doesn't expose the full redirect chain easily with head(),
            # so we do a second check: if final status is 2xx but there were redirects,
            # we check history.
            if resp.history:
                for hist_resp in resp.history:
                    dead, reason = is_dead_redirect(hist_resp)
                    if dead:
                        return True, reason
            return is_dead(resp)

        except RETRYABLE_EXCEPTIONS as e:
            if attempt == 1 and retry:
                time.sleep(1)  # brief pause before retry
                continue
            return True, f"{type(e).__name__}: {e}"

        except requests.RequestException as e:
            return True, f"{type(e).__name__}: {e}"


def is_dead_redirect(response: requests.Response) -> tuple[bool, str]:
    """Check an intermediate redirect response for dead-link signals."""
    status = response.status_code
    if status in REDIRECT_CODES:
        location = response.headers.get("Location", "")
        if location:
            loc_lower = location.lower()
            for pat in DEAD_LOCATION_PATTERNS:
                if pat in loc_lower:
                    return True, f"HTTP {status} → {location}"
    return False, ""


# ── Database helpers ────────────────────────────────────────────────────────

def get_feeds(db: sqlite3.Connection) -> dict[int, str]:
    """Return {feed_id: feed_name} mapping."""
    rows = db.execute("SELECT id, name FROM feed ORDER BY id").fetchall()
    return {row[0]: row[1] for row in rows}


def resolve_feed_ids(
    db: sqlite3.Connection,
    feed_ids: Optional[list[int]],
    feed_names: Optional[list[str]],
) -> set[int]:
    """Resolve feed filters into a set of feed IDs. Returns all if neither given."""
    if feed_ids:
        return set(feed_ids)

    feeds = get_feeds(db)
    if feed_names:
        name_lower_map = {name.lower(): fid for fid, name in feeds.items()}
        result = set()
        for name in feed_names:
            fid = name_lower_map.get(name.lower())
            if fid is not None:
                result.add(fid)
            else:
                # Try partial match
                matched = [
                    fid for fname, fid in name_lower_map.items()
                    if name.lower() in fname
                ]
                if matched:
                    result.update(matched)
                else:
                    print(f"⚠️  Feed '{name}' not found (available: {list(feeds.values())})",
                          file=sys.stderr)
        return result or set(feeds.keys())

    # No filter → all feeds
    return set(feeds.keys())


def fetch_entries(
    db: sqlite3.Connection,
    feed_ids: set[int],
    start_ts: Optional[int],
    end_ts: Optional[int],
) -> list[dict]:
    """
    Fetch entries matching the filters.

    Returns list of dicts: {id, title, link, date, id_feed, is_favorite, feed_name}
    """
    params = []
    conditions = []

    if feed_ids:
        placeholders = ",".join("?" for _ in feed_ids)
        conditions.append(f"e.id_feed IN ({placeholders})")
        params.extend(feed_ids)

    if start_ts is not None:
        conditions.append("e.date >= ?")
        params.append(start_ts)
    if end_ts is not None:
        conditions.append("e.date <= ?")
        params.append(end_ts)

    where = " AND ".join(conditions) if conditions else "1=1"

    query = f"""
        SELECT e.id, e.title, e.link, e.date, e.id_feed, e.is_favorite, f.name, e.content
        FROM entry e
        JOIN feed f ON e.id_feed = f.id
        WHERE {where}
        ORDER BY e.date DESC
    """

    rows = db.execute(query, params).fetchall()
    return [
        {
            "id": row[0],
            "title": row[1],
            "link": row[2],
            "date": row[3],
            "id_feed": row[4],
            "is_favorite": bool(row[5]),
            "feed_name": row[6],
            "content": row[7],
        }
        for row in rows
    ]


def get_or_create_tag(db_path: str, tag_name: str) -> int:
    """Get tag ID by name; create the tag if it doesn't exist. Returns tag ID."""
    conn = sqlite3.connect(db_path)
    tag_name = tag_name.strip()

    # Check if tag exists
    row = conn.execute(
        "SELECT id FROM tag WHERE name = ?", (tag_name,)
    ).fetchone()
    if row:
        conn.close()
        return row[0]

    # Create new tag
    conn.execute(
        "INSERT INTO tag(name, attributes) VALUES(?, '[]')",
        (tag_name,),
    )
    conn.commit()
    tag_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
    conn.close()
    return tag_id


def tag_entry(db_path: str, entry_id, tag_name: str) -> bool:
    """
    Add a user tag (我的标签) to an entry via the tag + entrytag tables.
    Uses INSERT OR IGNORE to safely skip duplicates.
    Returns True if a new row was inserted, False if already tagged.
    """
    tag_id = get_or_create_tag(db_path, tag_name)
    conn = sqlite3.connect(db_path)
    # SQLite: entrytag PK is (id_tag, id_entry), INSERT OR IGNORE skips duplicates
    cursor = conn.execute(
        "INSERT OR IGNORE INTO entrytag(id_tag, id_entry) VALUES(?, ?)",
        (tag_id, entry_id),
    )
    inserted = cursor.rowcount > 0
    conn.commit()
    conn.close()
    return inserted


def _extract_summary(content: Optional[str], max_chars: int = 300) -> str:
    """Extract a plain-text summary from FreshRSS entry content."""
    if not content:
        return ""

    # Extract the FULLCONTENT block if present
    start_marker = "<!-- FULLCONTENT start //-->"
    end_marker = "<!-- FULLCONTENT end //-->"
    if start_marker in content and end_marker in content:
        idx = content.index(start_marker) + len(start_marker)
        end_idx = content.index(end_marker)
        text = content[idx:end_idx]
    else:
        text = content

    # Strip HTML tags
    text = re.sub(r"<[^>]+>", "", text)
    # Decode common HTML entities
    text = text.replace("&nbsp;", " ")
    text = text.replace("&amp;", "&")
    text = text.replace("&lt;", "<")
    text = text.replace("&gt;", ">")
    text = text.replace("&quot;", '"')
    text = text.replace("&#39;", "'")
    text = text.replace("&ldquo;", "\u201c")
    text = text.replace("&rdquo;", "\u201d")
    # Collapse whitespace
    text = re.sub(r"[\u3000\t]+", "", text)
    text = re.sub(r"  +", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)
    text = text.strip()

    # Truncate to max_chars
    if len(text) > max_chars:
        text = text[:max_chars].rstrip() + "…"

    return text

# ── Output / reporting ──────────────────────────────────────────────────────

def format_date(ts: Optional[int]) -> str:
    """Format a Unix timestamp for display."""
    if ts is None:
        return "N/A"
    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M")


def terminal_report(results: list[dict]) -> None:
    """Print results as a formatted table to the terminal."""
    if not results:
        print("✅ No dead links found.")
        return

    # Get terminal width or default to 120
    try:
        term_width = os.get_terminal_size().columns
    except (OSError, ValueError):
        term_width = 120

    print(f"\n🔴 {len(results)} dead link(s) found:\n")

    # Dynamic column widths: Feed(18) Date(17) Status(30), rest → Title + URL
    feed_w = 18
    date_w = 17
    status_w = 30
    fixed = feed_w + date_w + status_w + 3  # +3 for spacing between cols
    remaining = max(term_width - fixed, 60)
    title_w = remaining // 3
    url_w = remaining - title_w - 1

    header = (f"{'Feed':<{feed_w}} {'Date':<{date_w}} {'Status':<{status_w}} "
              f"{'Title':<{title_w}} {'URL'}")
    print(header)
    print("─" * term_width)

    for r in results:
        feed = (r["feed_name"] or "(unknown)")[:feed_w]
        d = format_date(r["date"])
        reason = r["reason"][:status_w]
        title = (r["title"] or "(no title)")[:title_w]
        url = r["link"][:url_w]
        print(f"{feed:<{feed_w}} {d:<{date_w}} {reason:<{status_w}} "
              f"{title:<{title_w}} {url}")


def build_plaintext_report(results: list[dict]) -> str:
    """Build a plain-text email body — vertical layout, mobile-friendly."""
    if not results:
        return "FreshRSS Dead Link Scan: ✅ 未发现失效链接。"

    lines = [
        "FreshRSS Dead Link Scan Report",
        f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        f"Dead links found: {len(results)}",
        "",
    ]

    for i, r in enumerate(results):
        if i > 0:
            lines.append("***")
            lines.append("")
        lines.append(f"新闻源：{r['feed_name'] or '(unknown)'}")
        lines.append(f"日期时间：{format_date(r['date'])}")
        lines.append(f"失效原因：{r['reason']}")
        lines.append(f"新闻标题：{r['title'] or '(无标题)'}")
        lines.append(f"新闻链接：{r['link']}")
        summary = _extract_summary(r.get("content"), max_chars=350)
        if summary:
            lines.append(f"新闻摘要：\n {summary}")
        lines.append("")

    lines.append("─ End of report ─")
    return "\n".join(lines)


def send_email(
    smtp_host: str,
    smtp_port: int,
    smtp_user: str,
    smtp_pass: str,
    mail_from: str,
    mail_to: str,
    subject: str,
    body: str,
) -> None:
    """
    发送邮件，通过 _force_ipv4() 避免 IPv6 超时。

    端口: 465 = SSL, 587 = STARTTLS, 25 = 明文。
    """
    import ssl

    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = mail_from
    msg["To"] = mail_to
    msg.set_content(body)

    ctx = ssl.create_default_context()
    
    source_addr = ('0.0.0.0', 0)

    if smtp_port == 465:
        with smtplib.SMTP_SSL(smtp_host, smtp_port, timeout=30, context=ctx, source_address=source_addr) as server:
            server.login(smtp_user, smtp_pass)
            server.send_message(msg)
    elif smtp_port == 587:
        with smtplib.SMTP(smtp_host, smtp_port, timeout=30, source_address=source_addr) as server:
            server.ehlo()
            server.starttls(context=ctx)
            server.ehlo()
            server.login(smtp_user, smtp_pass)
            server.send_message(msg)
    else:
        with smtplib.SMTP(smtp_host, smtp_port, timeout=30, source_address=source_addr) as server:
            server.ehlo()
            server.login(smtp_user, smtp_pass)
            server.send_message(msg)

    print(f"📧 报告已发送至 {mail_to}")


# ── Main scanner ────────────────────────────────────────────────────────────

def scan(
    db_path: str,
    feed_ids: Optional[list[int]] = None,
    feed_names: Optional[list[str]] = None,
    start_date: Optional[str] = None,
    end_date: Optional[str] = None,
    days: Optional[int] = None,
    delay: float = 2.0,
    timeout: float = 15.0,
    max_redirects: int = 10,
    tag: Optional[str] = None,  # tag name to apply to dead entries (None = don't tag)
    star: bool = False,  # [Deprecated] use tag instead
    quiet: bool = False,
    smtp_host: Optional[str] = None,
    smtp_port: Optional[int] = None,
    smtp_user: Optional[str] = None,
    smtp_pass: Optional[str] = None,
    mail_from: Optional[str] = None,
    mail_to: Optional[str] = None,
    dry_run: bool = False,
) -> list[dict]:
    """
    Main scan routine.

    Returns list of dead-entry dicts (with extra 'reason' key).
    """
    # ── Handle deprecated --star ──
    if star and not tag:
        tag = "已删除"

    # ── Quiet-mode output gate ──
    def _echo(*args, **kwargs):
        """print() that respects --quiet."""
        if not quiet:
            print(*args, **kwargs)

    # ── Resolve date range ──
    now = datetime.now()
    start_ts: Optional[int] = None
    end_ts: Optional[int] = None

    if start_date:
        start_ts = int(datetime.strptime(start_date, "%Y-%m-%d").timestamp())
    if end_date:
        # End of day
        end_ts = int((datetime.strptime(end_date, "%Y-%m-%d") + timedelta(days=1)).timestamp()) - 1

    if days and not start_ts:
        start_ts = int((now - timedelta(days=days)).timestamp())
    if not end_ts:
        end_ts = int(now.timestamp())

    # ── Open DB (read-only for scan) ──
    db = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)

    # Resolve feeds
    target_feeds = resolve_feed_ids(db, feed_ids, feed_names)
    if not target_feeds:
        print("❌ No feeds matched your filter.", file=sys.stderr)
        db.close()
        return []

    feeds = get_feeds(db)
    feed_names_str = ", ".join(feeds[fid] for fid in target_feeds if fid in feeds)
    _echo(f"📡 扫描源: {feed_names_str}")
    _echo(f"📅 日期范围: {format_date(start_ts)} → {format_date(end_ts)}")
    _echo(f"⏱️  请求间隔: {delay}s")
    if dry_run:
        _echo("🧪 DRY RUN — 不会修改数据库")
    if tag:
        _echo(f"🏷️  标签: \"{tag}\"")

    # Fetch entries
    entries = fetch_entries(db, target_feeds, start_ts, end_ts)
    db.close()

    if not entries:
        _echo("未找到符合条件的条目。")
        return []

    _echo(f"📄 待检查条目: {len(entries)}\n")

    # ── Scan ──
    dead_results: list[dict] = []
    interrupted = False
    try:
        for i, entry in enumerate(entries, 1):
            url = entry["link"]
            if not url:
                continue

            _echo(f"[{i:4d}/{len(entries)}] {url[:100]}", end=" ", flush=True)
            is_dead_flag, reason = check_url(url, timeout, max_redirects)
            status_icon = "🔴" if is_dead_flag else "✅"
            _echo(status_icon, reason)

            if is_dead_flag:
                entry["reason"] = reason
                dead_results.append(entry)

            # Delay between requests (skip after the last one)
            if i < len(entries) and delay > 0:
                time.sleep(delay)
    except KeyboardInterrupt:
        interrupted = True
        _echo(f"\n\n⏸️  用户中断 — 已扫描 {len(dead_results)} 个失效链接（共检查 {i}/{len(entries)} 条）")

    # ── Tag dead URLs ──
    if tag and dead_results:
        if dry_run:
            _echo(f"\n🧪 DRY RUN — 将为 {len(dead_results)} 个条目打上标签 \"{tag}\"")
        else:
            _echo(f"\n🏷️  正在为 {len(dead_results)} 个失效条目打标签 \"{tag}\"...")
            tagged = 0
            for entry in dead_results:
                if tag_entry(db_path, entry["id"], tag):
                    tagged += 1
            _echo(f"完成（{tagged} 个新标记，{len(dead_results) - tagged} 个已有标签）。")

    # ── Resolve SMTP config: CLI args > file defaults ──
    _smtp_host = smtp_host or SMTP_CONFIG.get("host", "")
    _smtp_port = smtp_port if smtp_port is not None else SMTP_CONFIG.get("port", 587)
    _smtp_user = smtp_user or SMTP_CONFIG.get("user", "")
    _smtp_pass = smtp_pass or SMTP_CONFIG.get("pass", "")
    _mail_from = mail_from or SMTP_CONFIG.get("mail_from", "")
    if mail_to:
        _mail_to = mail_to
    else:
        cfg_to = SMTP_CONFIG.get("mail_to", "")
        _mail_to = cfg_to if cfg_to else None

    # ── Report ──
    email_configured = bool(_smtp_host and _smtp_user and _mail_from and _mail_to)

    if dead_results:
        # Build report once, use for both email and terminal
        report_subject = f"FreshRSS Dead Link Report — {len(dead_results)} dead"
        if interrupted:
            report_subject += " (interrupted)"

        if email_configured:
            body = build_plaintext_report(dead_results)
            if interrupted:
                body = f"[扫描被用户中断，以下为已获取的结果]\n\n{body}"
            send_email(
                smtp_host=_smtp_host,
                smtp_port=_smtp_port,
                smtp_user=_smtp_user,
                smtp_pass=_smtp_pass or "",
                mail_from=_mail_from,
                mail_to=_mail_to,
                subject=report_subject,
                body=body,
            )

        # Terminal output — skipped entirely when --quiet
        if not quiet:
            terminal_report(dead_results)
    else:
        _echo("\n✅ 未发现失效链接。")

    return dead_results


# ── CLI ─────────────────────────────────────────────────────────────────────

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="FreshRSS Dead Link Scanner",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent("""\
            Examples:
              %(prog)s freshrss.db --days 7
              %(prog)s freshrss.db --feed "竹新社" --start 2026-06-01 --tag "已删除"
              %(prog)s freshrss.db --days 30   # uses SMTP_CONFIG from file
        """),
    )

    # Required
    parser.add_argument("db", help="Path to the FreshRSS SQLite database")

    # Feed filters
    g_feed = parser.add_argument_group("Feed filter")
    g_feed.add_argument("--feed-id", type=int, action="append", dest="feed_ids",
                        help="Feed ID (can repeat)")
    g_feed.add_argument("--feed", type=str, action="append", dest="feed_names",
                        help="Feed name, case-insensitive substring match (can repeat)")

    # Date filters
    g_date = parser.add_argument_group("Date filter")
    g_date.add_argument("--days", type=int,
                        help="Scan entries from the last N days")
    g_date.add_argument("--start", dest="start_date", metavar="YYYY-MM-DD",
                        help="Start date (inclusive)")
    g_date.add_argument("--end", dest="end_date", metavar="YYYY-MM-DD",
                        help="End date (inclusive)")

    # Scanning
    g_scan = parser.add_argument_group("Scanning")
    g_scan.add_argument("--delay", type=float, default=2.0,
                        help="Delay in seconds between requests (default: 2.0)")
    g_scan.add_argument("--timeout", type=float, default=15.0,
                        help="HTTP request timeout in seconds (default: 15)")
    g_scan.add_argument("--max-redirects", type=int, default=10,
                        help="Maximum redirects to follow (default: 10)")

    # Actions
    g_act = parser.add_argument_group("Actions")
    g_act.add_argument("--tag", type=str, default=None, metavar="LABEL",
                       help="Tag dead URLs with a user label (我的标签), e.g. --tag \"已删除\"")
    g_act.add_argument("--quiet", action="store_true",
                       help="Suppress terminal report (use with SMTP to be silent)")
    g_act.add_argument("--star", action="store_true",
                       help="[Deprecated] Use --tag instead. Star (favorite) dead URLs in FreshRSS")
    g_act.add_argument("--dry-run", action="store_true",
                       help="Show what would be done without modifying the DB")

    # Email
    g_mail = parser.add_argument_group(
        "Email notification (SMTP)",
        "CLI args override SMTP_CONFIG in the script. "
        "Set SMTP_CONFIG in-file to avoid typing passwords on the command line.")
    g_mail.add_argument("--smtp-host", help="SMTP server hostname")
    g_mail.add_argument("--smtp-port", type=int,
                        help="SMTP port: 587=STARTTLS, 465=SSL, 25=plain "
                             "(default from file: 587)")
    g_mail.add_argument("--smtp-user", help="SMTP login username")
    g_mail.add_argument("--smtp-pass", help="SMTP login password (NOT recommended — "
                        "set SMTP_CONFIG['pass'] in the file instead)")
    g_mail.add_argument("--mail-from", help="From address")
    g_mail.add_argument("--mail-to", action="append", dest="mail_to",
                        help="To address (can repeat for multiple recipients)")

    return parser.parse_args()


def main() -> None:
    args = parse_args()

    mail_to = ", ".join(args.mail_to) if args.mail_to else None

    scan(
        db_path=args.db,
        feed_ids=args.feed_ids,
        feed_names=args.feed_names,
        start_date=args.start_date,
        end_date=args.end_date,
        days=args.days,
        delay=args.delay,
        timeout=args.timeout,
        max_redirects=args.max_redirects,
        star=args.star,
        tag=args.tag,
        quiet=args.quiet,
        dry_run=args.dry_run,
        smtp_host=args.smtp_host or None,
        smtp_port=args.smtp_port or None,
        smtp_user=args.smtp_user or None,
        smtp_pass=args.smtp_pass or None,
        mail_from=args.mail_from or None,
        mail_to=mail_to or None,
    )


if __name__ == "__main__":
    main()