ferrocarrilcentral/check.py

#!/usr/bin/python3

import configparser
import json
import logging
import smtplib
import sys
from datetime import date
from email.mime.text import MIMEText
from pathlib import Path

import lxml.html
import requests

# Spanish months with corresponding numbers (1-12)
spanish_months = {
    "enero": 1,
    "febrero": 2,
    "marzo": 3,
    "abril": 4,
    "mayo": 5,
    "junio": 6,
    "julio": 7,
    "agosto": 8,
    "septiembre": 9,
    "octubre": 10,
    "noviembre": 11,
    "diciembre": 12,
}

# Paths
CONFIG_DIR = Path.home() / ".config" / "ferrocarril"
CONFIG_FILE = CONFIG_DIR / "config"
DATA_DIR = Path.home() / "lib" / "data"
DATA_FILE = DATA_DIR / "ferrocarril_dates.json"
URL = "https://ferrocarrilcentral.com.pe/appfcca/"

# Configure logging
logger = logging.getLogger("FerrocarrilMonitor")
logger.setLevel(logging.INFO)

# Handler for stdout
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter("%(levelname)s: %(message)s")
handler.setFormatter(formatter)

# Only log INFO when in TTY, always log ERROR
if sys.stdout.isatty():
    logger.addHandler(handler)
else:
    handler.setLevel(logging.ERROR)
    logger.addHandler(handler)


def parse_dates(html_content: str) -> list[date]:
    doc = lxml.html.fromstring(html_content)
    dates = []

    for tour in doc.xpath('//div[contains(@class, "tourdate")]'):
        try:
            day = int(tour.xpath('.//div[contains(@class, "day")]')[0].text.strip())
            yeardate = tour.xpath('.//div[contains(@class, "yeardate")]')[0]
            month_str = (
                yeardate.xpath('.//div[contains(@class, "month")]')[0]
                .text.strip()
                .lower()
            )
            year = int(
                yeardate.xpath('.//div[contains(@class, "year")]')[0].text.strip()
            )

            month = spanish_months.get(month_str)
            if month is None:
                raise ValueError(f"Unknown month: {month_str}")

            dates.append(date(year, month, day))
        except (IndexError, ValueError) as e:
            logger.error(f"Error parsing date: {e}")
            continue

    return dates


def load_previous_dates() -> set[str]:
    """Load previously seen dates from file, return as set of ISO strings."""
    if DATA_FILE.exists():
        with open(DATA_FILE, "r") as f:
            return set(json.load(f))
    return set()


def save_dates(dates: list[date]) -> None:
    """Save current dates to file as ISO strings."""
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    with open(DATA_FILE, "w") as f:
        json.dump([d.isoformat() for d in dates], f)


def load_config() -> configparser.ConfigParser:
    """Load email configuration from INI file."""
    if not CONFIG_FILE.exists():
        raise FileNotFoundError(
            f"Config file not found at {CONFIG_FILE}. Please create it with SMTP details."
        )
    config = configparser.ConfigParser()
    config.read(CONFIG_FILE)
    if "mail" not in config:
        raise ValueError(f"Config file {CONFIG_FILE} must have an [mail] section")
    return config


def send_email(new_dates: list[date], config: configparser.ConfigParser) -> None:
    """Send email with new dates."""
    email_config = config["mail"]
    subject = "New Ferrocarril Central Dates Available!"
    body = (
        "New dates found:\n"
        + "\n".join(d.strftime("%d %B %Y") for d in new_dates)
        + f"\n\n{URL}"
    )

    msg = MIMEText(body)
    msg["Subject"] = subject
    msg["From"] = email_config["from_address"]
    msg["To"] = email_config["to_address"]

    try:
        with smtplib.SMTP(email_config["smtp_host"]) as server:
            server.send_message(msg)
        logger.info("Email sent successfully")
    except Exception as e:
        logger.error(f"Failed to send email: {e}")
        raise  # Re-raise to prevent saving dates if email fails


def main() -> None:
    # Ensure directories exist
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
    DATA_DIR.mkdir(parents=True, exist_ok=True)

    # Fetch webpage
    try:
        response = requests.get(URL, timeout=10)
        response.raise_for_status()
        html_content = response.text
    except requests.RequestException as e:
        logger.error(f"Failed to fetch webpage: {e}")
        return

    # Parse current dates
    current_dates = set(parse_dates(html_content))
    if not current_dates:
        logger.info("No dates found on webpage")
        return

    # Load previous dates
    previous_dates = load_previous_dates()

    # Check for new dates
    new_dates = [d for d in current_dates if d.isoformat() not in previous_dates]

    if new_dates:
        logger.info(f"New dates found: {new_dates}")
        try:
            config = load_config()
            send_email(new_dates, config)
            save_dates(
                list(current_dates)
            )  # Update stored dates only if email succeeds
        except FileNotFoundError as e:
            logger.error(str(e))
        except Exception as e:
            logger.error(f"Error in processing: {e}")
    else:
        logger.info("No new dates found")


if __name__ == "__main__":
    main()