#!/usr/bin/env python3 """ IVA Quarterly Filing Prep — finrecon skill. Queries CloudInvoice API for both HR (Quinta da Capelinha) and RMR entities, aggregates quarterly finalized documents, computes IVA (6% reduced rate for accommodation in mainland Portugal), and produces: 1. Markdown report at /docker/artifacts/finrecon/iva-YYYY-Qn.md 2. Telegram digest summary to João Usage: python3 iva-quarterly-prep.py [--quarter Q1|Q2|Q3|Q4] [--year 2026] Defaults to the most recently completed quarter. Environment: CLOUDINVOICE_HR_TOKEN — CloudInvoice API token for HR entity CLOUDINVOICE_RMR_TOKEN — CloudInvoice API token for RMR entity BOT_TOKEN — Telegram bot token (finrecon bot) ALERT_CHAT_ID — Telegram chat ID for João Exit codes: 0 — success 1 — API error or missing data 2 — missing env vars """ from __future__ import annotations import argparse import datetime import json import os import sys from collections import defaultdict from urllib.request import urlopen, Request from urllib.error import URLError, HTTPError # --- Constants --- CI_BASE = "https://api.cloudinvoice.net" IVA_RATE = 0.06 # Portugal mainland: IVA intermédio for accommodation ARTIFACTS_DIR = "/docker/artifacts/finrecon" ENTITIES = { "HR": { "env_token": "CLOUDINVOICE_HR_TOKEN", "label": "HR — Quinta da Capelinha", }, "RMR": { "env_token": "CLOUDINVOICE_RMR_TOKEN", "label": "RMR", }, } DOC_NATURE_MAP = { "FR": "Fatura-Recibo (FRE)", "FT": "Fatura (FAC)", "NC": "Nota de Crédito (NC)", "ND": "Nota de Débito (ND)", } QUARTER_MONTHS = { "Q1": (1, 3), "Q2": (4, 6), "Q3": (7, 9), "Q4": (10, 12), } def get_completed_quarter(now: datetime.date) -> tuple[int, str]: """Return (year, quarter_label) for the most recently completed quarter.""" m = now.month if m <= 1: return now.year - 1, "Q4" elif m <= 4: return now.year, "Q1" elif m <= 7: return now.year, "Q2" elif m <= 10: return now.year, "Q3" else: return now.year, "Q4" def quarter_date_range(year: int, q: str) -> tuple[str, str]: """Return (start_date, end_date) ISO strings for a quarter.""" start_m, end_m = QUARTER_MONTHS[q] start = f"{year}-{start_m:02d}-01" end_dt = datetime.date(year, end_m, 1) + datetime.timedelta(days=32) end_dt = end_dt.replace(day=1) - datetime.timedelta(days=1) end = end_dt.isoformat() return start, end def ci_get(token: str, endpoint: str) -> dict | None: """GET from CloudInvoice API. Returns parsed JSON or None on error.""" url = f"{CI_BASE}/{endpoint}" req = Request(url, headers={ "Authorization": f"Token {token}", "Content-Type": "application/json", "User-Agent": "FinRecon-IVA-Prep/1.0", }) try: with urlopen(req, timeout=30) as resp: return json.loads(resp.read()) except (URLError, HTTPError) as e: print(f" [WARN] CloudInvoice GET {endpoint} failed: {e}", file=sys.stderr) return None def fetch_quarter_docs(token: str, start: str, end: str) -> list[dict]: """Fetch all finalized documents in a date range, handling pagination.""" all_docs: list[dict] = [] offset = 0 max_pages = 50 # safety cap: 5000 docs max for _ in range(max_pages): data = ci_get(token, f"documents/?limit=100&offset={offset}&ordering=-id") if not data or "results" not in data: break batch = [ d for d in data["results"] if d.get("document_status", {}).get("code") == "F" and start <= d.get("document_date", "") <= end ] all_docs.extend(batch) # Stop if we've passed the date range or no more pages oldest = data["results"][-1].get("document_date", "9999") if data["results"] else "9999" if oldest < start or not data.get("next"): break offset += 100 return all_docs def compute_iva(gross: float) -> float: """Extract IVA from a gross (tax-inclusive) amount at 6%.""" return gross * IVA_RATE / (1 + IVA_RATE) def format_eur(v: float) -> str: return f"€{v:,.2f}" def safe_get(d: dict, *keys, default=0): """Nested safe get.""" for k in keys: if isinstance(d, dict): d = d.get(k, default) else: return default return d if d is not None else default # --- Main logic --- def process_entity(token: str, label: str, start: str, end: str) -> dict: """Fetch and aggregate docs for one entity.""" docs = fetch_quarter_docs(token, start, end) result = { "label": label, "total_docs": len(docs), "gross_total": 0.0, "iva_total": 0.0, "net_total": 0.0, "by_nature": defaultdict(lambda: {"count": 0, "gross": 0.0}), "by_month": defaultdict(lambda: {"count": 0, "gross": 0.0}), "by_method": defaultdict(lambda: {"count": 0, "gross": 0.0}), "docs": docs, } for doc in docs: gross = doc.get("total_amount", 0) or 0 nature_code = safe_get(doc, "document_nature", "code", default="??") nature_label = DOC_NATURE_MAP.get(nature_code, nature_code) month = doc.get("document_date", "")[:7] # YYYY-MM method = safe_get(doc, "payment_method", "description", default="Não definido") result["gross_total"] += gross result["by_nature"][nature_label]["count"] += 1 result["by_nature"][nature_label]["gross"] += gross result["by_month"][month]["count"] += 1 result["by_month"][month]["gross"] += gross result["by_method"][method]["count"] += 1 result["by_method"][method]["gross"] += gross # NC adjustments: subtract credit notes from gross nc_gross = sum(d.get("total_amount", 0) or 0 for d in docs if safe_get(d, "document_nature", "code") == "NC") net_revenue = result["gross_total"] - nc_gross result["iva_total"] = compute_iva(net_revenue) result["net_total"] = net_revenue - result["iva_total"] result["nc_gross"] = nc_gross return result def render_markdown(year: int, q: str, entities: dict[str, dict]) -> str: """Render the quarterly IVA report as markdown.""" start, end = quarter_date_range(year, q) lines = [ f"# IVA Quarterly Report — {q} {year}", f"", f"**Period:** {start} to {end}", f"**IVA Rate:** {IVA_RATE*100:.0f}% (IVA intermédio — alojamento, Portugal continental)", f"**Generated:** {datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}", f"", f"---", f"", ] combined_gross = 0.0 combined_iva = 0.0 combined_net = 0.0 for eid, edata in entities.items(): if not edata: continue lines.append(f"## {edata['label']}") lines.append(f"") lines.append(f"| Metric | Value |") lines.append(f"|---|---|") lines.append(f"| Total documents | {edata['total_docs']} |") lines.append(f"| Gross (c/ IVA) | {format_eur(edata['gross_total'])} |") lines.append(f"| Notas de crédito | {format_eur(edata['nc_gross'])} |") lines.append(f"| Receita líquida (base IVA) | {format_eur(edata['gross_total'] - edata['nc_gross'])} |") lines.append(f"| IVA liquidado ({IVA_RATE*100:.0f}%) | {format_eur(edata['iva_total'])} |") lines.append(f"| Líquido (s/ IVA) | {format_eur(edata['net_total'])} |") lines.append(f"") # By document nature if edata["by_nature"]: lines.append(f"### Por tipo de documento") lines.append(f"") lines.append(f"| Tipo | Qtd | Bruto |") lines.append(f"|---|---|---|") for nature, ndata in sorted(edata["by_nature"].items()): lines.append(f"| {nature} | {ndata['count']} | {format_eur(ndata['gross'])} |") lines.append(f"") # By month if edata["by_month"]: lines.append(f"### Por mês") lines.append(f"") lines.append(f"| Mês | Qtd | Bruto |") lines.append(f"|---|---|---|") for month, mdata in sorted(edata["by_month"].items()): lines.append(f"| {month} | {mdata['count']} | {format_eur(mdata['gross'])} |") lines.append(f"") # By payment method if edata["by_method"]: lines.append(f"### Por método de pagamento") lines.append(f"") lines.append(f"| Método | Qtd | Bruto |") lines.append(f"|---|---|---|") for method, mdata in sorted(edata["by_method"].items(), key=lambda x: -x[1]["gross"]): lines.append(f"| {method} | {mdata['count']} | {format_eur(mdata['gross'])} |") lines.append(f"") combined_gross += edata["gross_total"] combined_iva += edata["iva_total"] combined_net += edata["net_total"] # Combined summary lines.append(f"---") lines.append(f"") lines.append(f"## Resumo combinado (HR + RMR)") lines.append(f"") lines.append(f"| Métrica | Valor |") lines.append(f"|---|---|") lines.append(f"| Total bruto (c/ IVA) | {format_eur(combined_gross)} |") lines.append(f"| **IVA liquidado** | **{format_eur(combined_iva)}** |") lines.append(f"| Total líquido (s/ IVA) | {format_eur(combined_net)} |") lines.append(f"") lines.append(f"---") lines.append(f"") lines.append(f"**Nota para contabilista:** Este documento é gerado automaticamente a partir dos dados CloudInvoice.") lines.append(f"Verificar cruzamento com declaração periódica de IVA (Modelo 30).") lines.append(f"IVA liquidado calculado sobre receita líquida (excl. notas de crédito) à taxa de {IVA_RATE*100:.0f}%.") lines.append(f"") return "\n".join(lines) def send_telegram(bot_token: str, chat_id: str, text: str) -> bool: """Send message via Telegram Bot API.""" url = f"https://api.telegram.org/bot{bot_token}/sendMessage" payload = json.dumps({ "chat_id": chat_id, "text": text, "parse_mode": "Markdown", "disable_web_page_preview": True, }).encode() req = Request(url, data=payload, headers={ "Content-Type": "application/json", "User-Agent": "FinRecon-IVA-Prep/1.0", }) try: with urlopen(req, timeout=15) as resp: return resp.status == 200 except (URLError, HTTPError) as e: print(f" [WARN] Telegram send failed: {e}", file=sys.stderr) return False def build_telegram_digest(year: int, q: str, entities: dict[str, dict]) -> str: """Build a short Telegram digest.""" lines = [ f"📊 *IVA {q} {year} — Resumo para contabilista*", f"", ] combined_gross = 0.0 combined_iva = 0.0 for eid, edata in entities.items(): if not edata: continue nc_str = f" (NC: {format_eur(edata['nc_gross'])})" if edata["nc_gross"] > 0 else "" lines.append(f"*{edata['label']}:*") lines.append(f" Bruto: {format_eur(edata['gross_total'])}{nc_str}") lines.append(f" IVA ({IVA_RATE*100:.0f}%): {format_eur(edata['iva_total'])}") lines.append(f" Docs: {edata['total_docs']}") lines.append(f"") combined_gross += edata["gross_total"] combined_iva += edata["iva_total"] lines.append(f"*Total bruto:* {format_eur(combined_gross)}") lines.append(f"*Total IVA liquidado:* {format_eur(combined_iva)}") lines.append(f"") lines.append(f"Relatório completo: \`/docker/artifacts/finrecon/iva-{year}-{q}.md\`") return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description="IVA quarterly filing prep") parser.add_argument("--quarter", choices=["Q1", "Q2", "Q3", "Q4"], help="Quarter to process (default: most recently completed)") parser.add_argument("--year", type=int, help="Year (default: inferred from quarter)") parser.add_argument("--dry-run", action="store_true", help="Print report to stdout, don't write artifact or send Telegram") args = parser.parse_args() # Determine quarter now = datetime.date.today() if args.quarter and args.year: year, q = args.year, args.quarter elif args.quarter: year, _ = get_completed_quarter(now) q = args.quarter else: year, q = get_completed_quarter(now) start, end = quarter_date_range(year, q) print(f"IVA prep: {q} {year} ({start} → {end})") # Validate env missing = [] for eid, edata in ENTITIES.items(): if not os.environ.get(edata["env_token"]): missing.append(edata["env_token"]) if missing: print(f"[ERROR] Missing env vars: {', '.join(missing)}", file=sys.stderr) return 2 # Process each entity entities: dict[str, dict] = {} for eid, edata in ENTITIES.items(): token = os.environ[edata["env_token"]] print(f" Fetching {edata['label']}...") entities[eid] = process_entity(token, edata["label"], start, end) print(f" → {entities[eid]['total_docs']} docs, gross {format_eur(entities[eid]['gross_total'])}") # Render report report = render_markdown(year, q, entities) if args.dry_run: print("\n" + report) return 0 # Write artifact os.makedirs(ARTIFACTS_DIR, exist_ok=True) artifact_path = f"{ARTIFACTS_DIR}/iva-{year}-{q}.md" with open(artifact_path, "w") as f: f.write(report) print(f" Report written: {artifact_path}") # Send Telegram digest bot_token = os.environ.get("BOT_TOKEN", "") chat_id = os.environ.get("ALERT_CHAT_ID", "") if bot_token and chat_id: digest = build_telegram_digest(year, q, entities) if send_telegram(bot_token, chat_id, digest): print(" Telegram digest sent.") else: print(" [WARN] Telegram digest failed.", file=sys.stderr) else: print(" [INFO] BOT_TOKEN/ALERT_CHAT_ID not set — skipping Telegram.") return 0 if __name__ == "__main__": sys.exit(main())