A tua fatura é lida por uma IA?

"""Extração de dados de faturas de eletricidade — qualquer comercializador.

Não há fórmula/segredo nenhum de comercializador aqui: isto só lê os números
que já estão impressos na fatura (consumo, potência, preço atual). O layout
está calibrado e bem testado para a G9 (fatura real usada como referência),
mas os padrões são suficientemente genéricos para apanhar faturas de outros
comercializadores com rótulos parecidos. Frágil por natureza (regex sobre
texto extraído de PDF) — se os campos-chave não forem encontrados,
`missing_fields` fica preenchido e o chamador (main.py) mostra um formulário
manual de recurso em vez de um beco sem saída.
"""

from __future__ import annotations

import io
import logging
import re
from dataclasses import dataclass, field

import pdfplumber
import pytesseract

logger = logging.getLogger("electricity-advisor")

# Faturas geradas por alguns programas certificados (ex: EDP, "Contém
# Assinatura Digital") vêm sem nenhuma camada de texto — o texto é desenhado
# como vetor/curvas para fidelidade de impressão, não como carateres
# selecionáveis. pdfplumber devolve 0 carateres nesse caso; sem OCR, faturas
# assim iam sempre parar ao formulário manual, mesmo com uma fatura perfeita
# de um comercializador enorme. Limite de páginas para não deixar um PDF com
# centenas de páginas prender o processo em OCR (cada página demora ~1-2s).
_OCR_MIN_TEXT_LEN = 20
_OCR_MAX_PAGES = 4
_OCR_RESOLUTION = 200

# Gamas plausíveis para um perfil residencial em Portugal — última rede de
# segurança antes de aceitar um valor extraído (por regex OU por OCR) como
# verdadeiro. OCR pode ler "0,1671" como "1,671" (vírgula deslocada) ou outros
# erros de dígito; para uma ferramenta que informa uma decisão financeira,
# mais vale cair no formulário manual do que confiar num número implausível.
_PLAUSIBLE_RANGES: dict[str, tuple[float, float]] = {
    "monthly_kwh": (1.0, 5000.0),
    "contracted_power_kva": (1.0, 42.0),
    "current_energy_price_eur_kwh": (0.02, 0.60),
    "current_power_price_eur_dia": (0.02, 3.0),
}

REQUIRED_FIELDS = (
    "monthly_kwh",
    "contracted_power_kva",
    "opcao_horaria",
    "current_energy_price_eur_kwh",
    "current_power_price_eur_dia",
)

# Só para a frase-resumo ("o teu comercializador é a X") — não bloqueia nada
# se não encontrar; a comparação funciona na mesma sem saber o nome.
KNOWN_RETAILERS = [
    "G9", "EDP", "Endesa", "Galp", "Iberdrola", "Repsol", "Goldenergy",
    "MEO Energia", "SU Eletricidade", "Coopérnico", "Plenitude", "YEM",
    "Alfa Energia", "Fortia", "Audax",
]

# Palavras que sinalizam tarifário indexado ao mercado grossista — se
# nenhuma aparecer perto da tarifa, assume-se fixo (é o caso mais comum e é
# o que a maioria das faturas antigas/tradicionais tem).
INDEXED_KEYWORDS = ["indexad", "dinâmic", "dinamic", "OMIE", "mercado grossista", "mercado spot"]


@dataclass
class ParsedInvoice:
    monthly_kwh: float | None = None
    contracted_power_kva: float | None = None
    opcao_horaria: str | None = None
    current_energy_price_eur_kwh: float | None = None  # média ponderada quando `periods` está preenchido
    current_power_price_eur_dia: float | None = None
    # Consumo/preço por período ("vazio"/"ponta"/"cheia"/"fora_vazio") — só
    # preenchido quando os períodos têm preços DIFERENTES entre si (bi/tri-
    # horário a sério). Quando todos os períodos custam o mesmo, isso é
    # equivalente a Simples e fica None (nada a ganhar em separar).
    periods: dict[str, tuple[float, float]] | None = None
    retailer: str | None = None
    tariff_type: str = "fixo"  # "fixo" | "indexado" — heurística, ver _detect_tariff_type
    missing_fields: list[str] = field(default_factory=list)

    @property
    def is_usable(self) -> bool:
        return not any(f in self.missing_fields for f in REQUIRED_FIELDS)


def _pt_number(raw: str) -> float:
    """Converte um número em formato pt-PT ('1.234,56' ou '0,1348') para float."""
    return float(raw.replace(".", "").replace(",", "."))


def extract_text(file_bytes: bytes) -> str:
    with pdfplumber.open(io.BytesIO(file_bytes)) as pdf:
        text = "\n".join(page.extract_text() or "" for page in pdf.pages)
    if len(text.strip()) >= _OCR_MIN_TEXT_LEN:
        return text
    logger.warning("PDF sem camada de texto extraível, a tentar OCR")
    return _extract_text_ocr(file_bytes)


def _extract_text_ocr(file_bytes: bytes) -> str:
    with pdfplumber.open(io.BytesIO(file_bytes)) as pdf:
        parts = [
            pytesseract.image_to_string(page.to_image(resolution=_OCR_RESOLUTION).original, lang="por")
            for page in pdf.pages[:_OCR_MAX_PAGES]
        ]
    return "\n".join(parts)


def _detect_retailer(text: str) -> str | None:
    for name in KNOWN_RETAILERS:
        if re.search(re.escape(name), text, re.IGNORECASE):
            return name
    return None


def _detect_tariff_type(text: str) -> str:
    for kw in INDEXED_KEYWORDS:
        if re.search(kw, text, re.IGNORECASE):
            return "indexado"
    return "fixo"


# Formato tabular alternativo, visto em faturas emitidas por outro motor de
# faturação (ex: Goldenergy — "Microsoft Dynamics NAV-PT Localization"): sem
# as frases narrativas da G9 ("Potência Contratada (X kVA): Y €/dia"), em vez
# disso uma tabela com colunas Início/Fim/Qtd./Unid./Preço(€)/Valor(€)/IVA(%).
# Descoberto ao testar com uma fatura real da Goldenergy que os padrões da G9
# não apanhavam nada disto.

_POWER_ROW_RE = re.compile(
    r"Pot[êe]ncia Contratada\s+([\d.,]+)\s*kVA\s+\d{2}/\d{2}/\d{4}\s+\d{2}/\d{2}/\d{4}\s+\d+\s*Dia\s+([\d.,]+)",
    re.IGNORECASE,
)
# Nomes de período: Vazio/Ponta/Cheia (tri-horário) e Fora de Vazio/Fora Vazio
# (bi-horário), além de Simples — cobre os dois ciclos horários oficiais em
# Portugal, não só o tri-horário da fatura Goldenergy usada como referência.
_ENERGY_ROW_RE = re.compile(
    r"Consumo Eletricidade (Vazio|Ponta|Cheia|Fora\s*(?:de\s*)?Vazio|Simples) medido\s+"
    r"\d{2}/\d{2}/\d{4}\s+\d{2}/\d{2}/\d{4}\s+([\d.,]+)\s*kWh\s+([\d.,]+)",
    re.IGNORECASE,
)


def _normalize_period_name(raw: str) -> str:
    """"Vazio"/"vazio" -> "vazio"; "Fora Vazio"/"Fora de Vazio" -> "fora_vazio"."""
    normalized = re.sub(r"\s+", " ", raw.strip().lower())
    if "fora" in normalized:
        return "fora_vazio"
    return normalized


def _extract_power_tabular(text: str) -> tuple[float | None, float | None]:
    """(kVA, €/dia total) — pode haver duas linhas (acesso às redes +
    comercializador) para o mesmo escalão de potência; somam-se, tal como a
    G9 já soma isto numa única frase."""
    matches = _POWER_ROW_RE.findall(text)
    if not matches:
        return None, None
    kva = _pt_number(matches[0][0])
    total_price = sum(_pt_number(price) for _, price in matches)
    return kva, total_price


# Formato EDP (via OCR — ver extract_text): sem frases narrativas nem a
# tabela Início/Fim/Qtd./Unid. da Goldenergy. Em vez disso, uma linha com a
# opção horária ("Simples") seguida da linha "<intervalo de datas> N kWh
# PREÇO €", e uma linha "Potência (N kVA)" seguida de "<datas> N dias PREÇO
# €". Os nomes dos meses variam consoante o período faturado, por isso os
# padrões não os fixam — só a distância (poucos carateres) até ao número
# seguinte, para não apanhar sítios errados do documento (ex: "Simples"
# reaparece na frase-resumo da potência contratada e no parágrafo descritivo
# do consumo, sem "kWh" próximo, por isso essas ocorrências falham e o regex
# avança para a linha certa).
_EDP_ENERGY_RE = re.compile(r"\bSimples\b[\s\S]{0,60}?([\d.,]+)\s*k[Ww][Hh]\s+([\d.,]+)\s*€")
_EDP_POWER_RE = re.compile(r"Pot[êe]ncia\s*\(\s*([\d.,]+)\s*k[Vv][Aa]\)[\s\S]{0,60}?\d+\s*dias?\s+([\d.,]+)\s*€")


def _extract_energy_edp(text: str) -> tuple[float | None, float | None]:
    """(kWh, €/kWh) do formato EDP — só tarifário Simples (o único suportado
    nesta versão; Bi/Tri-horário teriam um rótulo diferente de "Simples")."""
    m = _EDP_ENERGY_RE.search(text)
    if not m:
        return None, None
    return _pt_number(m.group(1)), _pt_number(m.group(2))


def _extract_power_edp(text: str) -> tuple[float | None, float | None]:
    """(kVA, €/dia) do formato EDP."""
    m = _EDP_POWER_RE.search(text)
    if not m:
        return None, None
    return _pt_number(m.group(1)), _pt_number(m.group(2))


def _extract_energy_tabular(
    text: str,
) -> tuple[str | None, float | None, float | None, dict[str, tuple[float, float]] | None]:
    """(opcao_horaria, kWh total, €/kWh médio ponderado, detalhe por período)
    a partir de linhas "Consumo Eletricidade <Período> medido ... N kWh
    PREÇO". Um mesmo período pode aparecer em mais do que uma linha (faturas
    reais dividem por escalão de IVA — os primeiros 200 kWh/mês a 6%, o
    resto a 23% — sempre ao mesmo preço base); por isso soma-se o consumo
    por período antes de decidir se os preços diferem entre períodos.
    Períodos todos ao MESMO preço = efetivamente Simples (não há nada a
    ganhar em distinguir); preços diferentes = bi/tri-horário a sério, com
    `periods` preenchido para o cálculo exato do custo atual."""
    matches = _ENERGY_ROW_RE.findall(text)
    if not matches:
        return None, None, None, None

    by_period: dict[str, list[tuple[float, float]]] = {}
    for period, kwh_str, price_str in matches:
        by_period.setdefault(_normalize_period_name(period), []).append(
            (_pt_number(kwh_str), _pt_number(price_str))
        )

    periods: dict[str, tuple[float, float]] = {}
    for period_key, rows in by_period.items():
        period_kwh = sum(kwh for kwh, _ in rows)
        period_price = sum(kwh * price for kwh, price in rows) / period_kwh if period_kwh > 0 else rows[0][1]
        periods[period_key] = (period_kwh, period_price)

    total_kwh = sum(kwh for kwh, _ in periods.values())
    distinct_prices = {round(price, 6) for _, price in periods.values()}
    weighted_avg = sum(kwh * price for kwh, price in periods.values()) / total_kwh if total_kwh > 0 else None

    if len(distinct_prices) <= 1:
        # um único período (ex: "simples"), ou vários com o mesmo preço —
        # equivalente a Simples, sem necessidade de detalhe por período.
        return "simples", total_kwh, weighted_avg, None

    opcao = "bi-horario" if len(periods) == 2 else "tri-horario"
    return opcao, total_kwh, weighted_avg, periods


def parse_invoice(file_bytes: bytes) -> ParsedInvoice:
    text = extract_text(file_bytes)
    result = ParsedInvoice()
    result.retailer = _detect_retailer(text)
    result.tariff_type = _detect_tariff_type(text)

    # ── Potência ──
    # 1ª tentativa: "Potência Contratada (6,90 kVA): 0,4498 €/dia." (G9;
    # frase narrativa, rótulo comum a outros comercializadores).
    m = re.search(
        r"Pot[êe]ncia\s*Contratada\s*\(?\s*([\d.,]+)\s*kVA\)?\s*:?\s*([\d.,]+)?\s*€/dia",
        text,
        re.IGNORECASE,
    )
    if m and m.group(2):
        result.contracted_power_kva = _pt_number(m.group(1))
        result.current_power_price_eur_dia = _pt_number(m.group(2))
    else:
        # 2ª tentativa: formato tabular (ex: Goldenergy) — ver
        # _extract_power_tabular. Descoberto numa fatura real que não tinha
        # nenhuma frase narrativa, só linhas de tabela.
        kva, power_price = _extract_power_tabular(text)
        if kva is None or power_price is None:
            # 3ª tentativa: formato EDP (ver _extract_power_edp) — só chega
            # aqui via OCR, já que a fatura EDP real testada não tinha
            # camada de texto nenhuma.
            edp_kva, edp_power_price = _extract_power_edp(text)
            kva = edp_kva if kva is None else kva
            power_price = edp_power_price if power_price is None else power_price
        if kva is not None and power_price is not None:
            result.contracted_power_kva = kva
            result.current_power_price_eur_dia = power_price
        else:
            if kva is not None:
                result.contracted_power_kva = kva
            elif m and m.group(1):
                result.contracted_power_kva = _pt_number(m.group(1))
            else:
                result.missing_fields.append("contracted_power_kva")
            result.missing_fields.append("current_power_price_eur_dia")

    # ── Energia + opção horária + consumo ──
    # 1ª tentativa: "Energia Ativa (Simples): 0,1348 €/kWh." (G9; frase
    # narrativa — outros comercializadores costumam usar "Preço de
    # Energia"/"Tarifa" com o mesmo formato).
    m = re.search(
        r"(?:Energia\s*Ativa|Pre[çc]o\s*(?:de\s*)?Energia|Tarifa)\s*"
        r"\((Simples|Bi-?hor[aá]rio|Tri-?hor[aá]rio)\)\s*:?\s*([\d.,]+)\s*€/kWh",
        text,
        re.IGNORECASE,
    )
    if m:
        result.opcao_horaria = m.group(1).lower().replace("á", "a")
        result.current_energy_price_eur_kwh = _pt_number(m.group(2))
    else:
        # 2ª tentativa: formato tabular (linhas "Consumo Eletricidade
        # <Período> medido ... N kWh PREÇO") — dá logo opção horária,
        # consumo total, preço médio E detalhe por período, tudo de uma vez.
        opcao, total_kwh, price, periods = _extract_energy_tabular(text)
        if opcao is not None and price is not None:
            result.opcao_horaria = opcao
            result.current_energy_price_eur_kwh = price
            result.monthly_kwh = total_kwh
            result.periods = periods
        else:
            # 3ª tentativa: formato EDP (ver _extract_energy_edp).
            edp_kwh, edp_price = _extract_energy_edp(text)
            if edp_kwh is not None and edp_price is not None:
                result.opcao_horaria = "simples"
                result.current_energy_price_eur_kwh = edp_price
                result.monthly_kwh = edp_kwh
            else:
                result.missing_fields.append("opcao_horaria")
                result.missing_fields.append("current_energy_price_eur_kwh")

    # Se o consumo ainda não veio da extração tabular acima, tenta os
    # padrões narrativos da G9. Padrões específicos em alternativa, não um
    # único regex genérico à volta da palavra solta "consumo" — essa palavra
    # aparece várias vezes na fatura em contextos que nada têm a ver com o
    # total mensal (ex: nomes de linhas de imposto como "Imposto especial
    # consumo eletricidade"), e um regex demasiado permissivo apanhava o
    # "kWh" errado mais abaixo no documento.
    if result.monthly_kwh is None:
        consumo_patterns = [
            r"consumo\s+entre\b.*?foi\s*de\s*([\d.,]+)\s*kWh",  # "...consumo entre X e Y foi de N kWh" (G9)
            r"consumo\s+total\D{0,20}([\d.,]+)\s*kWh",  # "consumo total: N kWh"
            r"consumo\s*\(kWh\)\D{0,10}([\d.,]+)",  # "Consumo (kWh): N"
        ]
        m = None
        for pattern in consumo_patterns:
            m = re.search(pattern, text, re.IGNORECASE | re.DOTALL)
            if m:
                break
        if m:
            result.monthly_kwh = _pt_number(m.group(1))
        else:
            result.missing_fields.append("monthly_kwh")

    if result.opcao_horaria and result.opcao_horaria != "simples" and not result.periods:
        # Sabemos que não é Simples (ex: narrativa "Energia Ativa
        # (Bi-horário)") mas não conseguimos separar o consumo por período —
        # sem isso não dá para calcular o custo corretamente, e nunca
        # inventamos uma distribuição. Nunca visto ainda numa fatura real
        # (só o formato tabular, que já vem com o detalhe), mas mantido como
        # rede de segurança.
        result.missing_fields.append("opcao_horaria_unsupported")

    # Última rede de segurança: um valor fora de gama plausível (ex: OCR leu
    # "0,1671" como "1,671") vira "em falta" em vez de alimentar um veredito
    # financeiro errado com confiança total — ver _PLAUSIBLE_RANGES. Aplica-se
    # também a cada período individualmente, não só à média.
    for field_name, (lo, hi) in _PLAUSIBLE_RANGES.items():
        value = getattr(result, field_name)
        if value is not None and not (lo <= value <= hi):
            logger.warning("%s fora de gama plausível (%s), a tratar como em falta", field_name, value)
            setattr(result, field_name, None)
            result.periods = None
            if field_name not in result.missing_fields:
                result.missing_fields.append(field_name)

    if result.periods:
        lo, hi = _PLAUSIBLE_RANGES["current_energy_price_eur_kwh"]
        if any(not (lo <= price <= hi) for _, price in result.periods.values()):
            logger.warning("preço de um período fora de gama plausível, a descartar o detalhe por período")
            result.periods = None

    return result

"""Cliente mínimo do protocolo INSTREAM do clamd.

Evita depender do pyclamd (não mantido) — o protocolo é simples o suficiente
para implementar diretamente: enviar os bytes em chunks prefixados pelo
tamanho, terminar com um chunk vazio, ler a resposta. Qualquer falha de
ligação levanta ScanUnavailable — o chamador (main.py) trata isso como
"recusar o upload", nunca como "está limpo", porque um clamd em baixo não
pode ser motivo para saltar a verificação.
"""

from __future__ import annotations

import os
import socket
import struct

CLAMD_HOST = os.environ.get("CLAMD_HOST", "clamav")
CLAMD_PORT = int(os.environ.get("CLAMD_PORT", "3310"))
CHUNK_SIZE = 8192
TIMEOUT_SECONDS = 15


class ScanUnavailable(Exception):
    pass


def scan_bytes(data: bytes) -> tuple[bool, str | None]:
    """Devolve (limpo, nome_da_ameaça_ou_None). Levanta ScanUnavailable se o
    clamd não responder ou responder de forma inesperada."""
    try:
        with socket.create_connection((CLAMD_HOST, CLAMD_PORT), timeout=TIMEOUT_SECONDS) as sock:
            sock.sendall(b"zINSTREAM\0")
            for offset in range(0, len(data), CHUNK_SIZE):
                chunk = data[offset : offset + CHUNK_SIZE]
                sock.sendall(struct.pack("!L", len(chunk)) + chunk)
            sock.sendall(struct.pack("!L", 0))

            response = b""
            while True:
                part = sock.recv(4096)
                if not part:
                    break
                response += part
    except OSError as exc:
        raise ScanUnavailable(str(exc)) from exc

    text = response.decode("utf-8", errors="replace").strip("\0").strip()
    if text.endswith("OK"):
        return True, None
    if "FOUND" in text:
        threat = text.rsplit(":", 1)[-1].replace("FOUND", "").strip()
        return False, threat
    raise ScanUnavailable(f"resposta inesperada do clamd: {text!r}")