#!/usr/bin/env python3
"""allodi_scan.py - Allodi's free read-only scan of an Odoo database (Odoo 14 to 20).

WHAT IT DOES
  Reads Odoo's own metadata tables (installed modules, custom fields, views, actions, reports, scheduled jobs,
  access rules, company and user counts) with plain SELECT statements in a READ ONLY transaction, and writes:
    collect  ->  scan.json: counts plus names. Nothing else. This is the file a customer reviews and sends us.
    report   ->  classifies the modules, applies Allodi's published prices, and writes report.md, report.html and
                 appendix.md next to the scan.json (needs the module list odoo-modules.json or odoo-modules.js).
                 --issued YYYY-MM-DD is the report's date; the report itself never reads the clock.
    scan     ->  collect + report in one step.

WHAT IT NEVER DOES
  - It never imports or starts Odoo, so no scheduled job, email, automation or add-on code can run.
  - It never writes: every session is started READ ONLY (PGOPTIONS default_transaction_read_only=on, and each
    transaction is BEGIN ... READ ONLY ... ROLLBACK), with statement and lock timeouts.
  - It compares superuser powers before it reads. If your login has superuser powers and Odoo's own database user
    cannot act as a superuser, it stops and says so. Run it as the user Odoo runs as. --allow-superuser runs it
    anyway. It also stops, for any login, when a settings table it reads is not an ordinary table of Odoo's database
    user (or of a superuser, a role that can act as Odoo's user, or this login when it has no superuser powers). It
    then names the table. No flag overrides that.
  - It never reads business records: no partners, invoices, orders, messages, attachments, logins, passwords,
    email addresses, company names or code bodies. Tables read: ir_module_module, ir_model, ir_model_fields,
    ir_model_data, ir_ui_view, ir_act_server, base_automation, ir_cron, ir_act_report_xml, ir_model_access, ir_rule,
    ir_ui_menu, ir_act_window, website_page (view ids only), res_company and res_users (counts only), plus
    PostgreSQL's catalog (table and column names, who owns those tables and whether that owner and this login are
    superusers, database size, and the list of database names when you leave out --db). The exact queries are in
    this file.
  - It connects to your PostgreSQL server and to nothing else.

HOW TO RUN IT (customer, Path A)
  python3 allodi_scan.py collect
  or, to name the database yourself:
  python3 allodi_scan.py collect --db YOUR_DATABASE
  Uses psycopg2 when it is installed (Odoo's own Python has it), otherwise the psql command. Connection settings
  follow PostgreSQL's usual rules: --host/--port/--user, or the PGHOST, PGPORT, PGUSER, PGPASSWORD, PGDATABASE
  environment variables, or ~/.pgpass. On Odoo.sh, open the Shell tab and run it with no --db (PGDATABASE is set).
  It prints a summary and writes allodi-scan-<database>-<date>.json. Read it, then send it to us.

IF YOU DO NOT KNOW THE DATABASE NAME
  Leave out --db. The scan then opens one read-only session on the "postgres" database (or "template1") and reads
  the list of database names. It reads nothing else, and it opens no other database to look.
    One database:       it scans that one and tells you its name. ("postgres" is skipped when there are others.)
    Several:            it prints each name with the command to run for it, and stops. You choose.
    No list:            it tells you why, and how to see the names, and stops.
  The names are shown on your screen only. The scan file holds one database name: the one that was scanned. If
  PostgreSQL asks for a password, you type it once. With --db or PGDATABASE, the scan opens only that database.
  An empty --db '' (or PGDATABASE='') does not start this; it asks you to name the database.

LIMITS
  The scan trusts whoever owns Odoo's tables. That owner can change what the scan reads, so it can forge the counts
  and names. The owner can also hide code in its own tables, for example a column of its own type with a cast. That
  code runs inside the scan with the rights of the login that runs it. The scan's session is read-only, so that code
  cannot write data through it. It can still disturb that login's other sessions (for example end them, or keep the
  scan waiting until its time limit). Run as a superuser, that code would get superuser rights. That is why a
  superuser scan stops when the owner is not one.
  A settings table owned by another role is accepted only when that role can already act as Odoo's user, or is the
  login itself and the login has no superuser powers: its code then gains no rights it does not already have.
  Some superuser scans stop although the owner could reach superuser powers by a mixed route (on PostgreSQL 16 or
  later a chain of memberships mixing a SET-only link and an inherit-only link; on 12 to 15 an owner with CREATEROLE).
  That errs on the safe side; --allow-superuser runs it.

Python 3.8 or later, standard library only (psycopg2 optional). Exit status: 0 OK, 1 scan failed, 2 usage error.
"""
import argparse
import datetime
import html
import json
import math
import os
import re
import shlex
import shutil
import subprocess
import sys

TOOL_VERSION = "2026.10.03"
SCAN_FORMAT = 1
MODULE_ORIGIN_EXEMPT = ("__export__", "__import__", "studio_customization")   # xml ids that are NOT module code
INSTALLED_STATES = ("installed", "to upgrade", "to remove")
ODOO_REWRITTEN_VIEWS = (("base", "user_groups_view"),)   # views Odoo regenerates itself: never "edited standard views"
DEFAULT_TARGET = 19


# =====================================================================================================================
# Pricing: a line-by-line port of website/assets/pricing-core.js (quote, classification). test/parity_test.py runs
# the same cases through node and Python and fails on any difference, so keep the two in step.
# =====================================================================================================================
HOURLY_RATE = 150
PRICES = {
    # Halved 2026-10-24 (Mark): every monthly and one-time price; the hourly rate stays.
    "managed": {"firstCompany": 75, "extraCompany": 37.5, "branch": 12.5, "user": 15,
                "hoursPerCompany": 2, "hoursPerUser": 0.25, "hourly": HOURLY_RATE},
    "rescue": {"perDatabase": 750, "milestone": 250, "extraCompany": 125},
    "upgrade": {"moduleTiers": [[0, 0], [5, 750], [15, 2000], [40, 4500]], "overTopPerModule": 100,
                "perVersion": 500, "perEnterpriseApp": 250},
    "cleanup": {"tiers": [[50, 750], [150, 1750], [400, 3500]], "overTopPerItem": 7.5},
    # 24/7 outage response add-on, per month per Managed Odoo subscription (one database) (Mark, 2026-09-28, repriced
    # the same day): the greater of min or perUser x internal users (addon247_price). managed() adds it as one flat
    # line when the input has want247 (as pricing-core.js does).
    "addon247": {"min": 150, "perUser": 7.5},
}
MAX_COUNT = 100000
_JS_NUM = re.compile(r"^[+-]?(\d+\.?\d*|\.\d+)([eE][+-]?\d+)?$")


def js_number(v):
    """JavaScript Number(v) for the inputs the calculator sees (numbers, strings, booleans, null)."""
    if v is None:
        return 0.0
    if isinstance(v, bool):
        return 1.0 if v else 0.0
    if isinstance(v, (int, float)):
        return float(v)
    if isinstance(v, list):
        if not v:
            return 0.0
        return js_number(v[0]) if len(v) == 1 and not isinstance(v[0], list) else float("nan")
    if not isinstance(v, str):
        return float("nan")
    s = v.strip(" \t\n\r\x0b\x0c 

")
    if s == "":
        return 0.0
    if s in ("Infinity", "+Infinity"):
        return float("inf")
    if s == "-Infinity":
        return float("-inf")
    m = re.match(r"^0([xXoObB])([0-9a-fA-F]+)$", s)
    if m:
        base = {"x": 16, "o": 8, "b": 2}[m.group(1).lower()]
        try:
            return float(int(m.group(2), base))
        except ValueError:
            return float("nan")
    if _JS_NUM.match(s):
        return float(s)
    return float("nan")


def js_truthy(v):
    """JavaScript truthiness (if (v) ...) for JSON-like values: "", 0, NaN, null, false are false; [] and {} true."""
    if v is None or v is False:
        return False
    if isinstance(v, (int, float)) and not isinstance(v, bool):
        return v == v and v != 0
    if isinstance(v, str):
        return v != ""
    return True


def count(v, minimum=None):
    """Whole, non-negative number from any input ("", "3.7", -2, NaN -> 0), capped at MAX_COUNT."""
    x = js_number(v)
    n = math.floor(x) if math.isfinite(x) else None
    if n is None or n < 0:
        n = 0
    if n > MAX_COUNT:
        n = MAX_COUNT
    if minimum is not None and n < minimum:
        n = minimum
    return int(n)


def addon247_price(users):
    """Monthly price of the 24/7 outage response add-on: max(min, perUser x users). 10 -> 150, 20 -> 150, 25 -> 187.5,
    50 -> 375, 100 -> 750 (pricing-core.js addon247Price)."""
    a = PRICES["addon247"]
    return max(a["min"], a["perUser"] * count(users))


def managed(i):
    p = PRICES["managed"]
    companies = count(i.get("companies"), 1)
    branches = count(i.get("branches"))
    users = count(i.get("users"))
    lines = [
        {"label": "First company", "qty": 1, "unit": p["firstCompany"], "amount": p["firstCompany"]},
        {"label": "Additional companies", "qty": companies - 1, "unit": p["extraCompany"], "amount": (companies - 1) * p["extraCompany"]},
        {"label": "Branches", "qty": branches, "unit": p["branch"], "amount": branches * p["branch"]},
        {"label": "Internal users", "qty": users, "unit": p["user"], "amount": users * p["user"]},
    ]
    if js_truthy(i.get("want247")):
        # One line, qty 1 (the calculator hides qty-0 lines, and a 0-user subscription still pays the minimum).
        addon = addon247_price(users)
        lines.append({"label": "24/7 outage response", "qty": 1, "unit": addon, "amount": addon, "flat": True})
    return {"monthly": sum(l["amount"] for l in lines), "lines": lines,
            "includedHours": companies * p["hoursPerCompany"] + users * p["hoursPerUser"], "hourly": p["hourly"]}


def rescue(i):
    p = PRICES["rescue"]
    companies = count(i.get("companies"), 1)
    databases = min(count(i.get("databases"), 1), companies)
    extra = companies - databases
    extra_amount = extra * p["extraCompany"]
    start = databases * p["milestone"]
    cutover = databases * p["milestone"] + extra_amount
    handover = databases * p["milestone"]
    return {"databases": databases, "extraCompanies": extra, "total": start + cutover + handover, "milestones": [
        {"key": "start", "label": "At signing: server build and test copy restored", "amount": start},
        {"key": "cutover", "label": "At cutover: production live on your server", "amount": cutover},
        {"key": "handover", "label": "At handover: runbook, credentials, backups", "amount": handover}]}


def module_port_price(n):
    p = PRICES["upgrade"]
    n = count(n)
    for upper, price in p["moduleTiers"]:
        if n <= upper:
            return price
    top = p["moduleTiers"][-1]
    return top[1] + (n - top[0]) * p["overTopPerModule"]


def upgrade(i):
    p = PRICES["upgrade"]
    frm = count(i.get("fromVersion"))
    to = count(i.get("toVersion"))
    jumps = max(0, to - frm)
    modules = count(i.get("customModules"))
    apps = count(i.get("enterpriseApps"))
    lines = [
        {"label": "Custom modules to port", "qty": modules, "amount": module_port_price(modules)},
        {"label": "Major versions jumped", "qty": jumps, "unit": p["perVersion"], "amount": jumps * p["perVersion"]},
        {"label": "Enterprise apps replaced", "qty": apps, "unit": p["perEnterpriseApp"], "amount": apps * p["perEnterpriseApp"]},
    ]
    return {"total": lines[0]["amount"] + lines[1]["amount"] + lines[2]["amount"], "jumps": jumps, "lines": lines}


def cleanup(i):
    p = PRICES["cleanup"]
    n = count(i.get("customizations"))
    if n == 0:
        return {"total": 0, "items": 0}
    for upper, price in p["tiers"]:
        if n <= upper:
            return {"total": price, "items": n}
    top = p["tiers"][-1]
    return {"total": top[1] + (n - top[0]) * p["overTopPerItem"], "items": n}


def days_in_month(year, month):
    if month == 12:
        return 31
    return (datetime.date(year, month + 1, 1) - datetime.date(year, month, 1)).days


def js_round(x):
    return math.floor(x + 0.5)


def billing_day(date):
    """Day of the month Managed Odoo is billed on, from a 'YYYY-MM-DD' cutover date, or None (billingDay).
    No proration (Mark, 2026-09-26): the first month is billed in full on the start day."""
    m = re.match(r"^(\d{4})-(\d{2})-(\d{2})$", str(date or ""))
    if not m:
        return None
    y, mo, d = int(m.group(1)), int(m.group(2)), int(m.group(3))
    if y < 1 or mo < 1 or mo > 12:   # JS Date handles year 0; Python's date does not. Refused on both sides.
        return None
    if d < 1 or d > days_in_month(y, mo):
        return None
    return d


def quote(inp):
    m = managed(inp)
    r = rescue(inp) if inp.get("wantRescue") else None
    u = upgrade(inp) if inp.get("wantUpgrade") else None
    c = cleanup(inp) if inp.get("wantCleanup") else None
    if r:
        timeline = [
            {"key": "signing", "milestone": r["milestones"][0]["amount"]},
            {"key": "cutover", "milestone": r["milestones"][1]["amount"], "managedFirstMonth": m["monthly"],
             "billingDay": billing_day(inp.get("cutoverDate"))},
            {"key": "handover", "milestone": r["milestones"][2]["amount"]},
            {"key": "monthly", "monthly": m["monthly"]},
        ]
    else:
        timeline = [{"key": "monthly", "monthly": m["monthly"], "fromSignup": True}]
    one_time = (r["total"] if r else 0) + (u["total"] if u else 0) + (c["total"] if c else 0)
    return {"managed": m, "rescue": r, "upgrade": u, "cleanup": c, "timeline": timeline, "oneTime": one_time}


def money(n):
    """Whole dollars, or dollars and cents when there are cents: 1500 -> '$1,500', 37.5 -> '$37.50' (pricing-core.js money)."""
    c = js_round(abs(n) * 100)
    if c % 100:
        return ("-$" if n < 0 else "$") + f"{c // 100:,}.{c % 100:02d}"
    neg = n < 0
    n = abs(js_round(n))
    return ("-$" if neg else "$") + f"{n:,}"


def parse_version(v):
    """'saas~17.2' -> 17 (intermediate), '16.0+e' -> 16 (Enterprise build), '18.0' -> 18. Port of parseVersion."""
    if v is None:
        return None
    if isinstance(v, list):
        v = ".".join(str(x) for x in v[:2])
    s = str(v).strip()
    m = re.search(r"(?:saas[~-])?(\d{1,2})\.(\d{1,2})", s, re.I) or re.match(r"^(\d{1,2})()$", s)
    if not m:
        return None
    major = int(m.group(1))
    minor = int(m.group(2)) if m.group(2) else 0
    return {"raw": s, "major": major, "minor": minor,
            "intermediate": bool(re.search(r"saas", s, re.I)) or minor != 0,
            "enterprise": bool(re.search(r"\+e\b", s, re.I))}


L10N_ENTERPRISE_SUFFIXES = ["_reports", "_hr_payroll", "_hr_payroll_account", "_1099", "_payment_nacha",
                            "_intrastat", "_saft", "_reports_post_wizard", "_edi_extended", "_fec_import"]
NON_BILLABLE_APPS = {"web_mobile"}


def classify_module(name, D, target_bit, installed, source_bit, enterprise_db):
    """Port of classifyModule: community, enterprise, enterprise-likely, oca, custom (by name only)."""
    installed = installed or {}
    if name in D["community"]:
        mask = D["community"][name]
        return {"cls": "community", "onTarget": None if mask == 0 else bool(mask & target_bit)}
    in_oca = name in D["oca"]
    if name in D["enterprise"] and (enterprise_db or not (in_oca and source_bit and (D["oca"][name] & source_bit))):
        return {"cls": "enterprise", "app": name if name in D["enterpriseApps"] else None}
    if in_oca:
        return {"cls": "oca", "onTarget": bool(D["oca"][name] & target_bit) if target_bit else False}
    if re.search(r"_oca$", name):
        return {"cls": "oca", "onTarget": False, "guessed": True}
    if re.search(r"_enterprise$", name):
        return {"cls": "enterprise-likely", "reason": "name ends in _enterprise"}
    best = ""
    for e in D["enterprise"]:
        if len(e) <= len(best) or e not in installed:
            continue
        if len(e) >= 4 and name.startswith(e + "_"):
            best = e
            continue
        if len(name) > len(e) + 1 and name.endswith("_" + e):
            head = name[:-(len(e) + 1)]
            if head in D["community"] or head in D["enterprise"]:
                best = e
    if best:
        return {"cls": "enterprise-likely", "reason": "works with " + best}
    if name.startswith("l10n_"):
        for suf in L10N_ENTERPRISE_SUFFIXES:
            if len(name) > len(suf) and name.endswith(suf):
                return {"cls": "enterprise-likely", "reason": "Enterprise localization add-on"}
    return {"cls": "custom"}


def version_bit(D, major):
    key = f"{major}.0"
    return (1 << D["branches"].index(key)) if key in D["branches"] else 0


def analyze(parsed, D, target_major, overrides=None):
    """Port of analyze(): counts by class, modules to port, Enterprise apps in use (deduplicated by label)."""
    overrides = overrides or {}
    bit = version_bit(D, target_major)
    version = parse_version(parsed.get("version"))
    src_bit = version_bit(D, version["major"]) if version else 0
    ent_db = bool(version and version["enterprise"]) or ("web_enterprise" in parsed["modules"])
    names = sorted(parsed["modules"].keys())
    res = {"total": len(names), "rows": [], "counts": {"community": 0, "enterprise": 0, "enterprise-likely": 0, "oca": 0, "custom": 0},
           "toPort": 0, "ocaMissingOnTarget": 0, "apps": [], "version": version, "targetKnown": bit != 0,
           "ignored": parsed.get("ignored", 0)}
    seen_labels = set()
    for n in names:
        c = classify_module(n, D, bit, parsed["modules"], src_bit, ent_db)
        reviewable = c["cls"] in ("custom", "enterprise-likely") or (c["cls"] == "oca" and not c.get("onTarget"))
        ov = overrides[n] if reviewable and n in overrides else None
        row = {"name": n, "version": parsed["modules"][n], "cls": c["cls"], "reason": c.get("reason"),
               "onTarget": c.get("onTarget"), "override": ov, "ports": False, "reviewable": reviewable}
        res["counts"][c["cls"]] += 1
        effective = ov or ("custom" if c["cls"] == "custom" else "enterprise" if c["cls"] == "enterprise-likely" else None)
        if effective == "custom":
            row["ports"] = True
        if not ov and c["cls"] == "oca" and not c.get("onTarget"):
            row["ports"] = True
            res["ocaMissingOnTarget"] += 1
        if row["ports"]:
            res["toPort"] += 1
        if c["cls"] == "enterprise" and c.get("app") and c["app"] not in NON_BILLABLE_APPS:
            info = D["enterpriseApps"][c["app"]]
            if info["label"] not in seen_labels:
                seen_labels.add(info["label"])
                res["apps"].append({"name": c["app"], "label": info["label"], "note": info.get("note"), "oca": info.get("oca") or []})
        res["rows"].append(row)
    res["apps"].sort(key=lambda a: a["label"])
    return res


def load_module_data(path):
    """The module classification data: website/assets/odoo-modules.js, or infra's verify/odoo-modules.json."""
    with open(path, encoding="utf-8") as f:
        s = f.read()
    if path.endswith(".js"):
        s = s[s.index("=") + 1:].strip().rstrip(";")
    d = json.loads(s)
    apps = d.get("enterpriseAppsDetail") or d.get("enterpriseApps") or {}
    d["enterpriseApps"] = {k: (v if isinstance(v, dict) else {"label": v, "note": None, "oca": []}) for k, v in apps.items()}
    for k in ("branches", "community", "enterprise", "oca"):
        if k not in d:
            raise ValueError(f"{path}: not an Allodi module list (no '{k}')")
    return d


# =====================================================================================================================
# Read-only database access
# =====================================================================================================================
class ScanError(Exception):
    pass


class SessionRefused(ScanError):
    """The session check refused this login (superuser rule or table rule): nothing was read."""


# The session check fails with PostgreSQL's own error text, which carries one of these markers (any server language).
REFUSED_NOT_ODOO = "allodi_scan refused: not odoo"
REFUSED_SUPERUSER = "allodi_scan refused: superuser"
REFUSED_TABLE = "allodi_scan refused: table"          # followed by the table's name
# One line each, under 300 characters with "ERROR: " (the managed count keeps only the last line, cut at 300).
NOT_ODOO_MESSAGE = ("table ir_module_module not found: this does not look like an Odoo database (or this user cannot "
                    "read it)")
SUPERUSER_MESSAGE = ("This login is a PostgreSQL superuser and Odoo's database user is not, so it stops. Code hidden in "
                     "the database could run with full rights. Run it as Odoo's user (sudo -u odoo python3 "
                     "allodi_scan.py) or the read-only login from path_c_grants.sql. To accept that risk, add "
                     "--allow-superuser.")


def table_message(name):
    """The table rule's refusal, naming the table (always one of CATALOG_TABLES; shown through _printable)."""
    return (f"The scan stops because '{_printable(name)}' in this database is not a table that belongs to Odoo's "
            f"database user. Code in it could run with this login's rights. If it is Odoo's table, whoever manages "
            f"the database can fix this: change its owner to Odoo's database user.")


def _refusal(text):
    """SessionRefused when an error text carries a session-check marker, else None."""
    text = str(text)
    if REFUSED_NOT_ODOO in text:
        return SessionRefused(NOT_ODOO_MESSAGE)
    if REFUSED_SUPERUSER in text:
        return SessionRefused(SUPERUSER_MESSAGE)
    if REFUSED_TABLE in text:
        m = re.search(re.escape(REFUSED_TABLE) + r" ([A-Za-z0-9_]+)", text)
        name = m.group(1) if m and m.group(1) in CATALOG_TABLES else "a settings table"
        return SessionRefused(table_message(name))
    return None


def _wrap(sql):
    """One JSON text value per query: a list of row objects."""
    return f"SELECT COALESCE(pg_catalog.jsonb_agg(pg_catalog.to_jsonb(q)), '[]'::jsonb)::text FROM ({sql}) q"


class ReadOnlyDB:
    """Runs a batch of SELECT statements in ONE transaction that is READ ONLY at three levels: the session default
    (PGOPTIONS / options), the transaction itself (BEGIN ... READ ONLY), and ROLLBACK at the end."""

    def __init__(self, a):
        self.a = a
        self.timeout = f"{int(a.timeout)}s"
        # search_path is pg_catalog only (here and again with SET LOCAL in each transaction), and every function and
        # table is schema-qualified, so nothing created in "public" can stand in for what the scan calls.
        self.options = (f"-c default_transaction_read_only=on -c statement_timeout={int(a.timeout) * 1000} "
                        f"-c lock_timeout=5000 -c idle_in_transaction_session_timeout=600000 "
                        f"-c search_path=pg_catalog")
        self.backend = None
        if a.backend in ("auto", "psycopg2"):
            try:
                import psycopg2  # noqa: F401
                self.backend = "psycopg2"
            except ImportError:
                if a.backend == "psycopg2":
                    raise ScanError("psycopg2 is not installed; use --backend psql")
        if self.backend is None:
            if not shutil.which(a.psql or "psql"):
                raise ScanError("neither psycopg2 nor the psql command is available. Run this with Odoo's own Python "
                                "(it has psycopg2), or install the PostgreSQL client (psql).")
            self.backend = "psql"

    def _conn_kwargs(self):
        kw = {}
        for k in ("dbname", "host", "port", "user"):
            v = getattr(self.a, k if k != "dbname" else "db")
            if v:
                kw[k] = v
        if getattr(self.a, "password", None) is not None:   # typed once while reading the list (find_database)
            kw["password"] = self.a.password
        if getattr(self.a, "connect_timeout", None):         # reading the list only
            kw["connect_timeout"] = self.a.connect_timeout
        return kw

    def run(self, queries, check=True):
        """queries: list of (name, sql). Returns {name: [rows]}. check: run the session check first (every session
        except the one that reads the list of database names, which reads only pg_catalog.pg_database)."""
        if self.backend == "psycopg2":
            return self._run_psycopg2(queries, check)
        return self._run_psql(queries, check)

    def _check_sql(self):
        return session_check_sql(getattr(self.a, "allow_superuser", False))

    def _run_psycopg2(self, queries, check=True):
        import psycopg2
        kw = self._conn_kwargs()
        kw["options"] = self.options
        kw["application_name"] = "allodi_scan"
        try:
            conn = psycopg2.connect(**kw)
        except psycopg2.OperationalError as e:
            if _needs_password(e) and sys.stdin.isatty() and getattr(self.a, "ask_password", True):
                import getpass
                kw["password"] = getpass.getpass("PostgreSQL password: ")
                conn = psycopg2.connect(**kw)
            else:
                raise ScanError(f"cannot connect: {str(e).strip()}")
        out = {}
        try:
            conn.set_session(isolation_level="REPEATABLE READ", readonly=True, autocommit=False)
            cur = conn.cursor()
            cur.execute(SET_SEARCH_PATH)          # again here: a connection pooler can drop the startup options
            cur.execute("SET LOCAL statement_timeout = %s", (self.timeout,))
            cur.execute("SET LOCAL lock_timeout = '5s'")
            if check:
                self._check_psycopg2(cur)
            for name, sql in queries:
                try:
                    cur.execute(_wrap(sql))
                except psycopg2.Error as e:
                    raise ScanError(f"query '{name}' failed: {str(e).strip()}")
                out[name] = json.loads(cur.fetchone()[0])
            self.proof = self._session_proof_psycopg2(cur)
        finally:
            conn.rollback()
            conn.close()
        return out

    def _check_psycopg2(self, cur):
        import psycopg2
        try:
            cur.execute(self._check_sql())
        except psycopg2.Error as e:
            raise _refusal(e) or ScanError(f"session check failed: {str(e).strip()}")

    def _session_proof_psycopg2(self, cur):
        cur.execute("SELECT pg_catalog.current_setting('transaction_read_only'), "
                    "pg_catalog.current_setting('default_transaction_read_only')")
        tro, dtro = cur.fetchone()
        return {"transaction_read_only": tro, "default_transaction_read_only": dtro, "backend": "psycopg2"}

    def _psql_cmd(self):
        cmd = [self.a.psql or "psql", "-X", "-q", "-A", "-t", "-v", "ON_ERROR_STOP=1", "-f", "-"]
        db = self.a.db
        if db:
            # The database name is passed as a name and nothing else, as psycopg2 does. psql reads a bare -d value
            # that holds "=" or starts with postgresql:// as connection settings, which could point it elsewhere.
            db = "dbname='" + str(db).replace("\\", "\\\\").replace("'", "\\'") + "'"
        for flag, val in (("-d", db), ("-h", self.a.host), ("-p", self.a.port), ("-U", self.a.user)):
            if val:
                cmd += [flag, str(val)]
        if not getattr(self.a, "ask_password", True):
            cmd.append("-w")                                 # reading the list: psql does not stop to ask
        return cmd

    def _psql_env(self):
        env = dict(os.environ)
        env["PGOPTIONS"] = (env.get("PGOPTIONS", "") + " " + self.options).strip()
        env["PGAPPNAME"] = "allodi_scan"
        if getattr(self.a, "password", None) is not None:   # typed once while reading the list
            env["PGPASSWORD"] = self.a.password
        if getattr(self.a, "connect_timeout", None):         # reading the list only
            env["PGCONNECT_TIMEOUT"] = str(self.a.connect_timeout)
        return env

    def try_write(self, sql):
        """Self-test only: run one statement in a session opened like the scan's; return the error text or None."""
        if self.backend == "psycopg2":
            import psycopg2
            kw = self._conn_kwargs()
            kw["options"] = self.options
            kw["application_name"] = "allodi_scan_selftest"   # kept apart from the scan's own statement log
            try:
                conn = psycopg2.connect(**kw)
            except psycopg2.OperationalError as e:
                raise ScanError(f"cannot connect: {str(e).strip()}")
            try:
                conn.set_session(isolation_level="REPEATABLE READ", readonly=True, autocommit=False)
                cur = conn.cursor()
                cur.execute(SET_SEARCH_PATH)
                self._check_psycopg2(cur)            # a refusal is raised, not returned as the test's answer
                cur.execute(sql)
                return None
            except psycopg2.Error as e:
                return str(e)
            finally:
                conn.rollback()
                conn.close()
        script = ("\\set ON_ERROR_STOP 1\nBEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY;\n" + SET_SEARCH_PATH + ";\n"
                  + self._check_sql() + ";\n" + sql + ";\nROLLBACK;\n")
        env = self._psql_env()
        env["PGAPPNAME"] = "allodi_scan_selftest"
        p = subprocess.run(self._psql_cmd(), input=script, capture_output=True, text=True, env=env)
        if p.returncode != 0 and _refusal(p.stderr):
            raise _refusal(p.stderr)
        return None if p.returncode == 0 else (p.stderr or f"psql exit {p.returncode}")

    def _run_psql(self, queries, check=True):
        lines = ["\\set ON_ERROR_STOP 1", "BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY;", SET_SEARCH_PATH + ";",
                 f"SET LOCAL statement_timeout = '{self.timeout}';", "SET LOCAL lock_timeout = '5s';"]
        if check:
            lines.append(self._check_sql() + ";")         # its answer, 0, has no tab: the reader below skips it
        for name, sql in queries:
            if not re.match(r"^[a-z0-9_]+$", name):
                raise ScanError("internal: bad query name")
            lines.append(f"SELECT '{name}' || pg_catalog.chr(9) || ({_wrap(sql)});")
        lines.append("SELECT '__proof' || pg_catalog.chr(9) || pg_catalog.json_build_object('transaction_read_only', "
                     "pg_catalog.current_setting('transaction_read_only'), 'default_transaction_read_only', "
                     "pg_catalog.current_setting('default_transaction_read_only'))::text;")
        lines.append("ROLLBACK;")
        p = subprocess.run(self._psql_cmd(), input="\n".join(lines) + "\n", capture_output=True, text=True,
                           env=self._psql_env(), encoding="utf-8", errors="replace")
        if p.returncode != 0:
            raise _refusal(p.stderr) or ScanError("psql failed: " + (p.stderr.strip()[-800:] or f"exit {p.returncode}"))
        out = {}
        for line in p.stdout.splitlines():
            if "\t" not in line:
                continue
            name, payload = line.split("\t", 1)
            if name == "__proof":
                self.proof = dict(json.loads(payload), backend="psql")
            else:
                out[name] = json.loads(payload)
        missing = [n for n, _ in queries if n not in out]
        if missing:
            raise ScanError(f"psql returned no result for {missing[:3]}")
        return out


# =====================================================================================================================
# The queries
# =====================================================================================================================
CATALOG_TABLES = ("ir_module_module", "ir_model", "ir_model_fields", "ir_model_data", "ir_ui_view", "ir_act_server",
                  "base_automation", "ir_cron", "ir_act_report_xml", "ir_model_access", "ir_rule", "ir_ui_menu",
                  "ir_act_window", "website_page", "res_company", "res_users", "res_groups_users_rel",
                  "rel_server_actions", "studio_approval_rule", "ir_asset")
SET_SEARCH_PATH = "SET LOCAL search_path = pg_catalog"
# Membership in these lets a role run programs on the server or write its files, as a superuser can.
SERVER_PROGRAM_ROLES = ("pg_execute_server_program", "pg_write_server_files")
# The login test also counts reading the server's files: for the login, erring broad only refuses more.
LOGIN_POWER_ROLES = SERVER_PROGRAM_ROLES + ("pg_read_server_files",)
ODOO_OWNER = ("(SELECT k.relowner FROM pg_catalog.pg_class k "
              "WHERE k.oid = pg_catalog.to_regclass('public.ir_module_module'))")
# PostgreSQL 16 added the 'SET' privilege of pg_has_role and memberships WITH INHERIT FALSE, SET FALSE (which give
# nothing). The CASE evaluates 'SET' only on 16 or later; pg_has_role is STABLE and takes a column (s.oid), so the
# planner never evaluates a branch ahead of time (test_planted.py proves this with an unknown privilege name).
EFFECTIVE_FROM = 160000


def _role_list(names):
    return ", ".join(f"'{r}'" for r in names)


def _login_powers(role):
    """SQL, broad: true when role is a superuser or a member, by any route, of a superuser role or of
    pg_execute_server_program, pg_write_server_files or pg_read_server_files. Used for the scanning login only."""
    return (f"EXISTS (SELECT 1 FROM pg_catalog.pg_roles s WHERE (s.rolsuper OR s.rolname IN "
            f"({_role_list(LOGIN_POWER_ROLES)})) AND pg_catalog.pg_has_role({role}, s.oid, 'MEMBER'))")


def _owner_powers(role):
    """SQL, strict: true only when role can really act as a superuser: it is one, or it has the privileges of (or can
    SET ROLE to) a superuser role, pg_execute_server_program or pg_write_server_files. Before PostgreSQL 16 any member
    can SET ROLE, so 'MEMBER' is right there; from 16 a membership can give neither, so 'USAGE' or 'SET' is tested.
    Used for the owner of Odoo's tables and for the owner of each table the scan reads."""
    return (f"EXISTS (SELECT 1 FROM pg_catalog.pg_roles s WHERE (s.rolsuper OR s.rolname IN "
            f"({_role_list(SERVER_PROGRAM_ROLES)})) AND (CASE WHEN "
            f"pg_catalog.current_setting('server_version_num')::int >= {EFFECTIVE_FROM} THEN "
            f"pg_catalog.pg_has_role({role}, s.oid, 'USAGE') OR pg_catalog.pg_has_role({role}, s.oid, 'SET') "
            f"ELSE pg_catalog.pg_has_role({role}, s.oid, 'MEMBER') END))")


def _acts_as(role, target):
    """SQL: true when role can act as target, so it can already alter target's tables. From PostgreSQL 16 'USAGE'
    (role has target's privileges: inheriting them is what lets it alter target's tables); on 12 to 15 'MEMBER' (the
    documented right to SET ROLE to target). A SET-only route on 16 is not counted: refused, the safe direction."""
    return (f"(CASE WHEN pg_catalog.current_setting('server_version_num')::int >= {EFFECTIVE_FROM} THEN "
            f"pg_catalog.pg_has_role({role}, {target}, 'USAGE') "
            f"ELSE pg_catalog.pg_has_role({role}, {target}, 'MEMBER') END)")


def session_check_sql(allow_superuser=False):
    """The first statement of every scan session. It answers 0, or fails with PostgreSQL's own error text holding a
    REFUSED_* marker, so it is one plain SELECT on both backends (Path B's statement log accepts only plain reads).
    In this order:
    Not Odoo: there is no public.ir_module_module (any login): the usual "does not look like an Odoo database".
    Superuser rule: a login with superuser powers (broad test) stops when the owner of Odoo's tables cannot act as a
    superuser (strict test), because code that owner hid in the tables would run with more rights than it has.
    --allow-superuser lifts this rule only.
    Table rule (every scan, any login): it stops when a settings table the scan reads exists and is not an ordinary
    or partitioned table owned by one of: Odoo's owner; a role that can act as a superuser (strict test); a role that
    can act as Odoo's owner (_acts_as: it could already change Odoo's own tables, which the scan reads anyway); or the
    login itself, only when the login has no superuser powers (broad test), because its code then runs with rights
    it already has. Any other role could have made it, and its code would run with this login's rights. A view or
    any other kind of object under one of the names is refused whoever owns it. The answer names the first one."""
    scanner = _login_powers("session_user")
    login_oid = "(SELECT r.oid FROM pg_catalog.pg_roles r WHERE r.rolname = session_user)"
    odd_table = (f"(SELECT k.relname::text FROM pg_catalog.pg_class k JOIN pg_catalog.pg_namespace n ON "
                 f"n.oid = k.relnamespace WHERE n.nspname = 'public' AND k.relname IN ({_role_list(CATALOG_TABLES)}) "
                 f"AND (k.relkind NOT IN ('r', 'p') OR (k.relowner IS DISTINCT FROM {ODOO_OWNER} AND NOT "
                 f"{_owner_powers('k.relowner')} AND NOT {_acts_as('k.relowner', ODOO_OWNER)} AND NOT "
                 f"(k.relowner = {login_oid} AND NOT {scanner}))) ORDER BY k.relname LIMIT 1)")
    rules = (f"WHEN pg_catalog.to_regclass('public.ir_module_module') IS NULL THEN '{REFUSED_NOT_ODOO}' ")
    if not allow_superuser:
        rules += f"WHEN {scanner} AND NOT {_owner_powers(ODOO_OWNER)} THEN '{REFUSED_SUPERUSER}' "
    return (f"SELECT (CASE {rules}ELSE COALESCE('{REFUSED_TABLE} ' || {odd_table}, '0') END)::int AS allowed")


def q_catalog():
    tables = ", ".join(f"'{t}'" for t in CATALOG_TABLES)
    return [
        ("columns", f"SELECT table_name AS t, column_name AS c, data_type AS type FROM information_schema.columns "
                    f"WHERE table_schema = 'public' AND table_name IN ({tables})"),
        ("server", "SELECT pg_catalog.current_database() AS database, pg_catalog.current_setting('server_version') "
                   "AS pg_version, pg_catalog.pg_database_size(pg_catalog.current_database()) AS db_bytes, "
                   "current_user AS db_user, pg_catalog.now() AT TIME ZONE 'UTC' AS scanned_at_utc"),
    ]


class Cat:
    def __init__(self, rows):
        self.cols = {}
        for r in rows:
            self.cols.setdefault(r["t"], {})[r["c"]] = r["type"]

    def has(self, table, col=None):
        return table in self.cols and (col is None or col in self.cols[table])

    def text(self, table, alias, col):
        """A translatable column as English text (jsonb since Odoo 16, varchar before)."""
        t = self.cols.get(table, {}).get(col)
        if t is None:
            return "NULL::text"
        if t == "jsonb":
            return f"COALESCE({alias}.{col}->>'en_US', (SELECT value FROM pg_catalog.jsonb_each_text({alias}.{col}) LIMIT 1))"
        return f"{alias}.{col}::text"


def not_from_module(model, alias):
    """True when no INSTALLED module's data file defined the record (made in the interface, imported, exported,
    Studio, or left behind by a module that is no longer installed)."""
    ex = ", ".join(f"'{m}'" for m in MODULE_ORIGIN_EXEMPT)
    st = ", ".join(f"'{m}'" for m in INSTALLED_STATES)
    return (f"NOT EXISTS (SELECT 1 FROM public.ir_model_data d WHERE d.model = '{model}' AND d.res_id = {alias}.id "
            f"AND d.module NOT IN ({ex}) AND d.module IN (SELECT mm.name FROM public.ir_module_module mm WHERE mm.state IN ({st})))")


def from_module(model, alias):
    return "NOT (" + not_from_module(model, alias) + ")"


def studio_mark(model, alias):
    return (f"EXISTS (SELECT 1 FROM public.ir_model_data d WHERE d.model = '{model}' AND d.res_id = {alias}.id "
            f"AND d.module = 'studio_customization')")


def q_main(cat):
    Q = []
    mod_cols = ["name", "state", "latest_version"]
    for c in ("author", "license", "application"):
        if cat.has("ir_module_module", c):
            mod_cols.append(c)
    Q.append(("modules", f"SELECT {', '.join('m.' + c for c in mod_cols)} FROM public.ir_module_module m "
                         f"WHERE m.state IN ('installed', 'to upgrade', 'to remove', 'to install') ORDER BY m.name"))

    # Companies and branches: counts only (no names).
    active = "c.active" if cat.has("res_company", "active") else "TRUE"
    parent = "c.parent_id IS NOT NULL" if cat.has("res_company", "parent_id") else "FALSE"
    Q.append(("companies", f"SELECT pg_catalog.count(*) FILTER (WHERE {active} AND NOT ({parent})) AS top_level, "
                           f"pg_catalog.count(*) FILTER (WHERE {active} AND {parent}) AS with_parent, "
                           f"pg_catalog.count(*) FILTER (WHERE NOT {active}) AS archived FROM public.res_company c"))

    # Users: counts only. Internal = active and not share (has the Internal User group). Portal = active and share,
    # minus members of the Public group (the website visitor account) where the group table is readable.
    public_filter = ""
    if cat.has("res_groups_users_rel", "uid") and cat.has("res_groups_users_rel", "gid"):
        public_filter = (" AND NOT EXISTS (SELECT 1 FROM public.res_groups_users_rel r JOIN public.ir_model_data d ON "
                         "d.model = 'res.groups' AND d.module = 'base' AND d.name = 'group_public' AND d.res_id = r.gid "
                         "WHERE r.uid = u.id)")
    Q.append(("users", f"SELECT pg_catalog.count(*) FILTER (WHERE u.active AND NOT u.share) AS internal, "
                       f"pg_catalog.count(*) FILTER (WHERE u.active AND u.share{public_filter}) AS portal, "
                       f"pg_catalog.count(*) FILTER (WHERE NOT u.active) AS archived FROM public.res_users u"))

    # Custom models and fields (state = 'manual': made in the interface, by Studio or by import; never module code).
    # A model made with Studio counts ONCE (Mark, 2026-09-26): the fields, views, access rights and record rules on it
    # are its scaffolding and are not counted again; their numbers are reported with the model.
    sm = (f"(SELECT sm.model FROM public.ir_model sm WHERE sm.state = 'manual' AND {studio_mark('ir.model', 'sm')} "
          f"AND {not_from_module('ir.model', 'sm')})")
    sm_ids = (f"(SELECT sm.id FROM public.ir_model sm WHERE sm.state = 'manual' AND {studio_mark('ir.model', 'sm')} "
              f"AND {not_from_module('ir.model', 'sm')})")
    Q.append(("custom_models", f"SELECT m.id, m.model, {cat.text('ir_model', 'm', 'name')} AS label, "
                               f"{studio_mark('ir.model', 'm')} AS studio FROM public.ir_model m WHERE m.state = 'manual' "
                               f"AND {not_from_module('ir.model', 'm')} ORDER BY m.model"))
    # Odoo 17+ creates one manual column per extra analytic plan (x_plan<N>_id, a many2one to account.analytic.account)
    # on the analytic models. Odoo makes and maintains them itself, so they are not customizations.
    analytic = "(f.name ~ '^x_plan[0-9]+_id$' AND f.relation = 'account.analytic.account')"
    Q.append(("custom_fields", f"SELECT f.id, f.model, f.name, f.ttype, "
                               f"{cat.text('ir_model_fields', 'f', 'field_description')} AS label, "
                               f"(f.name LIKE 'x\\_studio\\_%' OR {studio_mark('ir.model.fields', 'f')}) AS studio, "
                               f"EXISTS (SELECT 1 FROM public.ir_model mm WHERE mm.model = f.model AND mm.state = 'manual') AS on_custom_model "
                               f"FROM public.ir_model_fields f WHERE f.state = 'manual' AND NOT {analytic} "
                               f"AND {not_from_module('ir.model.fields', 'f')} AND f.model NOT IN {sm} ORDER BY f.model, f.name"))
    Q.append(("analytic_plan_columns", f"SELECT pg_catalog.count(*) AS n FROM public.ir_model_fields f WHERE f.state = 'manual' AND {analytic}"))

    # Views. Website content (pages and the per-website copies the website builder makes) is not a customization.
    website_excl = []
    if cat.has("ir_ui_view", "website_id"):
        website_excl.append("v.website_id IS NULL")
    if cat.has("website_page", "view_id"):
        website_excl.append("NOT EXISTS (SELECT 1 FROM public.website_page wp WHERE wp.view_id = v.id)")
    not_website = " AND ".join(website_excl) or "TRUE"
    is_website = f"NOT ({not_website})"
    inherit = "v.inherit_id IS NOT NULL" if cat.has("ir_ui_view", "inherit_id") else "FALSE"
    key = "v.key" if cat.has("ir_ui_view", "key") else "NULL::text"
    Q.append(("custom_views", f"SELECT v.id, v.name, v.model, v.type, {key} AS key, {inherit} AS inherited, v.active, "
                              f"{studio_mark('ir.ui.view', 'v')} AS studio FROM public.ir_ui_view v "
                              f"WHERE {not_from_module('ir.ui.view', 'v')} AND {not_website} "
                              f"AND (v.model IS NULL OR v.model NOT IN {sm}) ORDER BY v.model, v.name, v.id"))
    if cat.has("ir_ui_view", "arch_updated"):
        # Views Odoo itself rewrites at runtime (so arch_updated is set without anyone editing them): the user form's
        # access-rights section (base.user_groups_view, regenerated whenever groups change, Odoo 14-18).
        auto = " OR ".join(f"(d.module = '{m}' AND d.name = '{n}')" for m, n in ODOO_REWRITTEN_VIEWS)
        rewritten = (f"EXISTS (SELECT 1 FROM public.ir_model_data d WHERE d.model = 'ir.ui.view' AND d.res_id = v.id AND ({auto}))")
        Q.append(("edited_views", f"SELECT v.id, v.name, v.model, v.type, {key} AS key, v.active FROM public.ir_ui_view v "
                                  f"WHERE v.arch_updated AND {from_module('ir.ui.view', 'v')} AND {not_website} "
                                  f"AND NOT {rewritten} ORDER BY v.model, v.name, v.id"))
        Q.append(("rewritten_views", f"SELECT pg_catalog.count(*) AS n FROM public.ir_ui_view v WHERE v.arch_updated AND {rewritten}"))
    Q.append(("website_views", f"SELECT pg_catalog.count(*) AS n FROM public.ir_ui_view v WHERE {is_website}"))

    # Server actions made in the interface (usage = ir_actions_server: not the hidden action behind a scheduled job
    # or an automation rule, which are counted there). Code is never read, only its length.
    usage = "s.usage = 'ir_actions_server'" if cat.has("ir_act_server", "usage") else "TRUE"
    if cat.has("base_automation", "action_server_id"):   # Odoo 14-16: an automation IS a server action
        usage += " AND NOT EXISTS (SELECT 1 FROM public.base_automation ba WHERE ba.action_server_id = s.id)"
    if cat.has("ir_act_server", "base_automation_id"):   # Odoo 17+: the actions of an automation rule
        usage += " AND s.base_automation_id IS NULL"
    if cat.has("ir_cron", "ir_actions_server_id"):
        usage += " AND NOT EXISTS (SELECT 1 FROM public.ir_cron cr WHERE cr.ir_actions_server_id = s.id)"
    if cat.has("rel_server_actions", "action_id") and cat.has("rel_server_actions", "server_id"):
        # the children of a "multi" action that belongs to an automation rule or a scheduled job belong to that rule/job
        owners = []
        if cat.has("base_automation", "action_server_id"):
            owners.append("SELECT action_server_id FROM public.base_automation")
        if cat.has("ir_act_server", "base_automation_id"):
            owners.append("SELECT id FROM public.ir_act_server WHERE base_automation_id IS NOT NULL")
        if cat.has("ir_cron", "ir_actions_server_id"):
            owners.append("SELECT ir_actions_server_id FROM public.ir_cron")
        if owners:
            usage += (" AND NOT EXISTS (SELECT 1 FROM public.rel_server_actions rs WHERE rs.action_id = s.id AND rs.server_id IN ("
                      + " UNION ".join(owners) + "))")
    # Path C roles may be granted every column except the code itself (it can hold hard-coded secrets): then the
    # length is simply not reported. information_schema only lists columns this role may read.
    code_len = "COALESCE(pg_catalog.length(s.code), 0)" if cat.has("ir_act_server", "code") else "NULL::int"
    Q.append(("server_actions", f"SELECT s.id, {cat.text('ir_act_server', 's', 'name')} AS name, s.state, im.model, "
                                f"{code_len} AS code_chars, {studio_mark('ir.actions.server', 's')} AS studio "
                                f"FROM public.ir_act_server s LEFT JOIN public.ir_model im ON im.id = s.model_id "
                                f"WHERE {usage} AND {not_from_module('ir.actions.server', 's')} ORDER BY im.model, s.id"))

    # Automation rules (automated actions).
    if cat.has("base_automation"):
        if cat.has("base_automation", "action_server_id"):     # 14-16
            Q.append(("automations", f"SELECT a.id, {cat.text('ir_act_server', 's', 'name')} AS name, im.model, a.trigger, "
                                     f"a.active, s.state AS action_types, (s.state = 'code') AS runs_code "
                                     f"FROM public.base_automation a JOIN public.ir_act_server s ON s.id = a.action_server_id "
                                     f"LEFT JOIN public.ir_model im ON im.id = s.model_id "
                                     f"WHERE {not_from_module('base.automation', 'a')} ORDER BY im.model, a.id"))
        else:                                                   # 17+
            link = "s.base_automation_id = a.id" if cat.has("ir_act_server", "base_automation_id") else "FALSE"
            Q.append(("automations", f"SELECT a.id, {cat.text('base_automation', 'a', 'name')} AS name, im.model, a.trigger, "
                                     f"a.active, (SELECT pg_catalog.string_agg(DISTINCT s.state, ', ') FROM public.ir_act_server s WHERE {link}) AS action_types, "
                                     f"EXISTS (SELECT 1 FROM public.ir_act_server s WHERE {link} AND s.state = 'code') AS runs_code "
                                     f"FROM public.base_automation a LEFT JOIN public.ir_model im ON im.id = a.model_id "
                                     f"WHERE {not_from_module('base.automation', 'a')} ORDER BY im.model, a.id"))

    # Scheduled jobs.
    if cat.has("ir_cron", "cron_name"):
        cname = cat.text("ir_cron", "c", "cron_name")
        cjoin = ""
    else:
        cname = cat.text("ir_act_server", "cs", "name")
        cjoin = "LEFT JOIN public.ir_act_server cs ON cs.id = c.ir_actions_server_id"
    cmodel = ("(SELECT im.model FROM public.ir_act_server x JOIN public.ir_model im ON im.id = x.model_id WHERE x.id = c.ir_actions_server_id)"
              if cat.has("ir_cron", "ir_actions_server_id") else "NULL::text")
    Q.append(("scheduled_jobs", f"SELECT c.id, {cname} AS name, c.active, c.interval_number, c.interval_type, {cmodel} AS model "
                                f"FROM public.ir_cron c {cjoin} WHERE {not_from_module('ir.cron', 'c')} ORDER BY c.id"))

    # Reports.
    Q.append(("reports", f"SELECT r.id, {cat.text('ir_act_report_xml', 'r', 'name')} AS name, r.model, r.report_name, "
                         f"r.report_type, {studio_mark('ir.actions.report', 'r')} AS studio FROM public.ir_act_report_xml r "
                         f"WHERE {not_from_module('ir.actions.report', 'r')} ORDER BY r.model, r.id"))

    # Access rights and record rules (SOW Appendix E). Names and models only; rule domains are never read.
    acc_active = "a.active" if cat.has("ir_model_access", "active") else "TRUE"
    Q.append(("access_rights", f"SELECT a.id, a.name, im.model, {acc_active} AS active FROM public.ir_model_access a "
                               f"LEFT JOIN public.ir_model im ON im.id = a.model_id WHERE {not_from_module('ir.model.access', 'a')} "
                               f"AND (a.model_id IS NULL OR a.model_id NOT IN {sm_ids}) ORDER BY im.model, a.id"))
    Q.append(("record_rules", f"SELECT r.id, r.name, im.model, r.active FROM public.ir_rule r LEFT JOIN public.ir_model im ON im.id = r.model_id "
                              f"WHERE {not_from_module('ir.rule', 'r')} AND (r.model_id IS NULL OR r.model_id NOT IN {sm_ids}) "
                              f"ORDER BY im.model, r.id"))
    # What each Studio model brought with it (reported beside the model, never counted).
    Q.append(("studio_scaffolding", f"SELECT m.model, "
                                    f"(SELECT pg_catalog.count(*) FROM public.ir_model_fields f WHERE f.model = m.model AND f.state = 'manual') AS fields, "
                                    f"(SELECT pg_catalog.count(*) FROM public.ir_ui_view v WHERE v.model = m.model AND {not_from_module('ir.ui.view', 'v')}) AS views, "
                                    f"(SELECT pg_catalog.count(*) FROM public.ir_model_access a WHERE a.model_id = m.id AND {not_from_module('ir.model.access', 'a')}) AS access_rights, "
                                    f"(SELECT pg_catalog.count(*) FROM public.ir_rule r WHERE r.model_id = m.id AND {not_from_module('ir.rule', 'r')}) AS record_rules "
                                    f"FROM public.ir_model m WHERE m.id IN {sm_ids} ORDER BY m.model"))

    # Studio approval rules (Enterprise) and asset files (custom JavaScript/CSS bundles entries, Odoo 15+).
    if cat.has("studio_approval_rule"):
        aname = cat.text("studio_approval_rule", "x", "name") if cat.has("studio_approval_rule", "name") else "NULL::text"
        amodel = "(SELECT im.model FROM public.ir_model im WHERE im.id = x.model_id)" if cat.has("studio_approval_rule", "model_id") else "NULL::text"
        aact = "x.active" if cat.has("studio_approval_rule", "active") else "TRUE"
        Q.append(("approval_rules", f"SELECT x.id, {aname} AS name, {amodel} AS model, {aact} AS active FROM public.studio_approval_rule x "
                                    f"WHERE {not_from_module('studio.approval.rule', 'x')} ORDER BY x.id"))
    if cat.has("ir_asset"):
        wsa = "AND x.website_id IS NULL" if cat.has("ir_asset", "website_id") else ""
        Q.append(("assets", f"SELECT x.id, x.name, x.bundle, x.path, x.active FROM public.ir_asset x "
                            f"WHERE {not_from_module('ir.asset', 'x')} {wsa} ORDER BY x.id"))

    # Counted but NOT part of the customization total (listed under "not counted").
    Q.append(("ui_menus", f"SELECT pg_catalog.count(*) AS n FROM public.ir_ui_menu x WHERE {not_from_module('ir.ui.menu', 'x')}"))
    Q.append(("ui_window_actions", f"SELECT pg_catalog.count(*) AS n FROM public.ir_act_window x WHERE {not_from_module('ir.actions.act_window', 'x')}"))
    return Q


# =====================================================================================================================
# collect
# =====================================================================================================================
CATEGORIES = [
    # key, label, what it means
    ("custom_models", "Custom models", "Record types created in the interface or with Studio. A model made with Studio counts once, with the fields, views, access rights and record rules on it."),
    ("custom_fields", "Custom fields", "Fields created in the interface or with Studio, including Studio fields on standard models (not those of a Studio model, counted with it)."),
    ("custom_views", "Custom and inherited views", "Views that no installed module defines: made in developer mode, with Studio, or imported."),
    ("edited_views", "Edited standard views", "Views that come from a module but whose layout was changed in the database."),
    ("server_actions", "Server actions", "Actions made in the interface; those that run Python code are marked."),
    ("automations", "Automation rules", "Automated actions made in the interface (one per rule, whatever its actions)."),
    ("scheduled_jobs", "Scheduled jobs", "Scheduled actions (cron jobs) made in the interface."),
    ("reports", "Reports", "Report actions made in the interface or with Studio (their templates are counted as views)."),
    ("access_rights", "Access rights", "Model access lines made in the interface."),
    ("record_rules", "Record rules", "Record rules made in the interface."),
    ("approval_rules", "Approval rules", "Studio approval rules on buttons (Enterprise)."),
    ("assets", "Asset files", "JavaScript or style files added to Odoo's bundles in the database (Odoo 15 and later)."),
]


def collect(a):
    db = ReadOnlyDB(a)
    first = db.run(q_catalog())
    cat = Cat(first["columns"])
    for t in ("ir_module_module", "ir_model_fields", "ir_ui_view", "res_users", "res_company"):
        if not cat.has(t):
            raise ScanError(f"table {t} not found: this does not look like an Odoo database (or this user cannot read it)")
    res = db.run(q_main(cat))
    server = first["server"][0]
    mods = res["modules"]
    installed = [m for m in mods if m["state"] in INSTALLED_STATES]
    base = next((m for m in mods if m["name"] == "base"), None)
    comp = res["companies"][0]
    users = res["users"][0]
    base_version = base["latest_version"] if base else None
    v = parse_version(base_version)
    major = v["major"] if v else None
    branches_apply = bool(major and major >= 17)
    cust = {k: res.get(k, []) for k, _, _ in CATEGORIES}
    scaf = {x["model"]: {k: int(x[k]) for k in ("fields", "views", "access_rights", "record_rules")} for x in res.get("studio_scaffolding", [])}
    for m in cust["custom_models"]:
        if m.get("studio"):
            m["scaffolding"] = scaf.get(m["model"], {"fields": 0, "views": 0, "access_rights": 0, "record_rules": 0})
    scan = {
        "format": SCAN_FORMAT,
        "tool": f"allodi_scan.py {TOOL_VERSION}",
        "scanned_at_utc": str(server["scanned_at_utc"])[:19] + "Z",
        "database": server["database"],
        "postgres": {"version": server["pg_version"], "database_bytes": server["db_bytes"]},
        "read_only_proof": dict(db.proof, db_user=server["db_user"]),
        "odoo": {"base_version": base_version, "major": major, "intermediate": bool(v and v["intermediate"]),
                 "web_enterprise_installed": any(m["name"] == "web_enterprise" for m in installed)},
        "companies": {
            "companies": int(comp["top_level"]) + (0 if branches_apply else int(comp["with_parent"])),
            "branches": int(comp["with_parent"]) if branches_apply else 0,
            "archived": int(comp["archived"]),
            "rule": "active companies without a parent; from Odoo 17, active companies with a parent are branches",
        },
        "users": {"internal": int(users["internal"]), "portal": int(users["portal"]), "archived": int(users["archived"]),
                  "rule": "internal = active and not share; portal = active and share, excluding the Public group"},
        "modules": [{k: m.get(k) for k in ("name", "state", "latest_version", "author", "license", "application")} for m in mods],
        "customizations": {k: cust[k] for k, _, _ in CATEGORIES},
        "not_counted": {
            "website_views": int(res["website_views"][0]["n"]),
            "menus_made_in_interface": int(res["ui_menus"][0]["n"]),
            "window_actions_made_in_interface": int(res["ui_window_actions"][0]["n"]),
            "analytic_plan_columns": int(res["analytic_plan_columns"][0]["n"]),
            "views_odoo_rewrites": int(res["rewritten_views"][0]["n"]) if "rewritten_views" in res else 0,
        },
        "filestore": None,
    }
    if a.filestore:
        scan["filestore"] = filestore_size(a.filestore)
    return scan


def filestore_size(path):
    """File count and bytes of an Odoo filestore folder. Reads directory entries only, never file contents."""
    n = size = 0
    for root, _dirs, files in os.walk(path):
        if os.path.basename(root) == "checklist":
            continue
        for fn in files:
            try:
                size += os.path.getsize(os.path.join(root, fn))
                n += 1
            except OSError:
                pass
    return {"path_given": True, "files": n, "bytes": size}


# =====================================================================================================================
# report: classification, quote, plain-English output
# =====================================================================================================================
def _is_oca(author):
    return "odoo community association" in author.lower()


def _by_author(author, lic):
    """custom or thirdparty from the author a module declares (docs/16-free-scan.md 2.2)."""
    if _is_oca(author):
        return "oca", "author OCA, not in the OCA list for this version"
    if author and author.lower() not in ("odoo s.a.", "odoo sa", "odoo s.a"):
        return "thirdparty", ("paid Apps Store license (OPL-1)" if lic == "OPL-1" else f"author {author}")
    return "custom", ("no author" if not author else "author left at the Odoo default")


def module_origin(m, cls):
    """Finer label from the database's own module metadata (author, license). Documented in docs/16-free-scan.md.
    Returns (origin, why, confirm): confirm=True means Mark checks it before the quote is final."""
    author = (m.get("author") or "").strip()
    lic = (m.get("license") or "").strip()
    if cls == "custom" and lic == "OEEL-1" and "odoo s.a" in author.lower():
        return "enterprise", "license OEEL-1 by Odoo S.A., not in the published Enterprise list", True
    if cls == "enterprise-likely":
        if lic == "OEEL-1":
            return "enterprise", "name and license OEEL-1 of an Odoo Enterprise add-on", False
        origin, why = _by_author(author, lic)                 # a name like Enterprise's, but not Odoo's license
        return origin, f"Enterprise-like name without license OEEL-1; {why}", False
    if cls == "oca" and author and not _is_oca(author):
        return "thirdparty", f"OCA module name, but author {author}: a vendor copy", False
    if cls == "custom":
        o, w = _by_author(author, lic)
        return o, w, False
    return cls, None, False


# ---------------------------------------------------------------------------------------------------------------------
# Dates. The report never reads the clock: the same scan.json with the same arguments always gives the same report,
# byte for byte (the website keeps a sample and compares it; hq files a report once per fingerprint of report.html).
# The report's date is the --issued date its caller passes. Without one it is the scan's own date, when that date is
# believable. A scan date that is missing, not a real date, or out of range makes the report preliminary.
# ---------------------------------------------------------------------------------------------------------------------
QUOTE_VALID_DAYS = 30                         # Mark, 2026-10-01: the quote holds for 30 days from the report's date
REPORT_WORDING = "2026-10-01"                 # the date of the report's wording, written to report.json
SCAN_TOOL_SINCE = datetime.date(2026, 9, 1)   # this tool was first released in September 2026: no scan is older
ISSUED_LATEST = datetime.date(2200, 1, 1)     # --issued: a typing slip like 20261-... or 9999 is refused
SCAN_AFTER_ISSUE_DAYS = 1                     # a scan may be dated up to a day after its report (time zones)
# With no --issued there is no "today" to compare the scan's date with, and the report does not read the clock. The
# module list's own date is the one clock-free yardstick: the list is refreshed far more often than every 400 days,
# so a scan dated later than that after it is not believable.
SCAN_AFTER_MODULE_LIST_DAYS = 400
OLD_SCAN_DAYS = 30                            # an older scan is named in the report (the fees bind on the scan date)
MONTH_NAMES = ("January", "February", "March", "April", "May", "June", "July", "August", "September", "October",
               "November", "December")
_SCAN_TIME = re.compile(r"^(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2})(?::(\d{2})(?:\.\d{1,9})?)?(Z|[+-]\d{2}:?\d{2})?\Z")


def parse_date(value):
    """'2026-09-30' -> datetime.date(2026, 9, 30). Anything else (another format, February 30, not text) -> None."""
    m = re.match(r"^(\d{4})-(\d{2})-(\d{2})\Z", value) if isinstance(value, str) else None
    if not m:
        return None
    try:
        return datetime.date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
    except ValueError:
        return None


def parse_scan_time(value):
    """A scan's scanned_at_utc -> (date, 'HH:MM UTC' or None), or None when it is not a real date and time.
    'Z' is UTC. Another offset ('+02:00') is converted to UTC, so the date and the time are right. A time with no
    zone keeps its date and shows no time: it is not known to be UTC, so it is never labelled UTC."""
    m = _SCAN_TIME.match(value) if isinstance(value, str) else None
    if not m:
        return None
    try:
        t = datetime.datetime(int(m.group(1)), int(m.group(2)), int(m.group(3)), int(m.group(4)), int(m.group(5)),
                              int(m.group(6) or 0))
        zone = m.group(7)
        if zone is None:
            return t.date(), None
        if zone != "Z":
            hours, minutes = int(zone[1:3]), int(zone[-2:])
            if hours > 14 or minutes > 59:
                return None
            shift = datetime.timedelta(hours=hours, minutes=minutes)
            t = t - shift if zone[0] == "+" else t + shift
    except (ValueError, OverflowError):
        return None
    return t.date(), f"{t.hour:02d}:{t.minute:02d} UTC"


def long_date(d):
    """datetime.date -> 'September 30, 2026'. Month names are fixed here, not taken from the computer's language
    settings, so a report reads the same wherever it is made."""
    return f"{MONTH_NAMES[d.month - 1]} {d.day}, {d.year}"


def module_data_date(D):
    """The date the module list was generated, or None."""
    m = re.search(r"\d{4}-\d{2}-\d{2}", str(D.get("generated") or D.get("source") or ""))
    return parse_date(m.group(0)) if m else None


def plural(n, one, many=None):
    """'1 Enterprise app', '6 Enterprise apps'."""
    return f"{n} {one if n == 1 else (many or one + 's')}"


def build_report(scan, D, target=DEFAULT_TARGET, overrides=None, issued=None):
    """issued: the report's date (a datetime.date, from --issued), or None for the scan's own date."""
    overrides = dict(overrides or {})
    mods = {m["name"]: m for m in scan["modules"] if m["state"] in INSTALLED_STATES}
    pending = sorted(m["name"] for m in scan["modules"] if m["state"] not in INSTALLED_STATES)
    excluded = []
    if "studio_customization" in mods:
        excluded.append("studio_customization")   # Studio's container: its contents are counted as customizations
        mods.pop("studio_customization")
    v = parse_version(scan["odoo"]["base_version"])
    if v and v["major"] < 1:
        v = None            # "0.0" and the like is not an Odoo version: unreadable, in every statement of the report
    parsed = {"modules": {n: m.get("latest_version") for n, m in mods.items()},
              "version": scan["odoo"]["base_version"] if v else None, "ignored": 0}
    # "Before this quote is final": points confirmed with the customer before the quote is binding. Each point is
    # kept twice, in the same order: `checks` is the internal copy, worded for Allodi's staff (the contact intake
    # puts it in the lead's note; never printed in the report), and `checks_customer` is what the report prints,
    # written to the customer: what is confirmed and how it can change a price.
    checks = []
    checks_customer = []

    def check(staff, customer):
        checks.append(staff)
        checks_customer.append(customer)

    major = v["major"] if v else None
    tgt = target
    tgt_note = None
    if v and v["intermediate"] and major >= target:
        tgt = major + 1
        tgt_note = (f"Your database is on intermediate version {v['raw']}; it can only move to {tgt}.0 or later, "
                    f"after Odoo Online upgrades it.")
    if not version_bit(D, tgt):
        check(f"The module list has no Odoo {tgt}.0 yet: which standard and OCA modules exist on {tgt}.0 is not "
              f"known, so every OCA module is counted to port. Confirm against {tgt}.0 before the quote is final.",
              f"Our module list does not cover Odoo {tgt}.0 yet, so this report counts every OCA module as a module "
              f"to port. We will check your modules against Odoo {tgt}.0 and confirm the count with you. OCA modules "
              f"already released for Odoo {tgt}.0 come off the count, which can lower the upgrade price.")
    if major and not version_bit(D, major):
        # The module list (D["branches"]) has no data for the database's own version: standard, Enterprise and OCA
        # modules of that version may be sorted wrongly, so the counts are confirmed before anything is binding.
        known = sorted(int(b.split(".")[0]) for b in D["branches"] if re.match(r"^\d+\.", b))
        rel = "newer than" if known and major > known[-1] else "older than" if known and major < known[0] else "not covered by"
        span = f"{known[0]}.0 to {known[-1]}.0" if known else "no versions"
        check(f"Odoo {major} is {rel} the module list ({span}): modules of "
              f"that version may be classified wrongly. Confirm the module and Enterprise app counts with the customer.",
              f"Your database is on Odoo {major}, which is {rel} the module list this report uses, so module and "
              f"Enterprise app counts will be confirmed with you first. A different count can change the project fees.")
    # Automatic corrections from the database's metadata, applied the way the pricing page applies a visitor's
    # correction (only rows the page lets a visitor correct); explicit --override values win.
    pre = analyze(parsed, D, tgt, {})
    auto = {}
    for row in pre["rows"]:
        origin, _why, _c = module_origin(mods[row["name"]], row["cls"])
        if row["cls"] == "custom" and origin == "enterprise":
            auto[row["name"]] = "enterprise"
        if row["cls"] == "enterprise-likely" and origin != "enterprise":
            auto[row["name"]] = "custom"
    eff_over = dict(auto)
    eff_over.update(overrides)
    a = analyze(parsed, D, tgt, eff_over)
    rows = []
    to_port = a["toPort"]
    for row in a["rows"]:
        m = mods[row["name"]]
        origin, why, confirm = module_origin(m, row["cls"])
        ov = row.get("override")
        if ov and row["name"] in overrides:
            origin = {"enterprise": "enterprise", "thirdparty": "vendor_supplied", "drop": "dropped",
                      "custom": origin if origin in ("custom", "thirdparty") else "custom"}[ov]
            why = f"set on review: {ov}"
            confirm = False
        if row["cls"] == "oca" and origin == "thirdparty" and not row["ports"] and not ov:
            row = dict(row, ports=True)                        # an OCA name under a vendor's author: port it
            to_port += 1
        if confirm:
            check(f"Module {row['name']}: license OEEL-1 by Odoo S.A. but not in the published Enterprise list. "
                  f"Counted as Odoo Enterprise (not ported); confirm it is Odoo's own and not a custom module.",
                  f"Module {row['name']} states Odoo's Enterprise license (OEEL-1) and Odoo S.A. as its author, but it "
                  f"is not in our list of Odoo Enterprise modules. This report treats it as part of Odoo Enterprise "
                  f"and does not count it to port. We will confirm with you where it came from: if it was written "
                  f"for you, it adds one module to port, which can raise the upgrade price.")
        rows.append(dict(row, origin=origin, why=why, author=m.get("author"), license=m.get("license"),
                         application=bool(m.get("application"))))
    by_origin = {}
    for r in rows:
        by_origin.setdefault(r["origin"], []).append(r)

    apps = list(a["apps"])
    listed = {x["name"] for x in apps}
    for r in rows:
        if (r["origin"] == "enterprise" and r["application"] and r["name"] not in D["enterpriseApps"]
                and r["name"] not in NON_BILLABLE_APPS and r["name"] not in listed):
            apps.append({"name": r["name"], "label": r["name"], "oca": [], "unlisted": True,
                         "note": "Not in the published replacement list; the replacement is agreed with you."})
            check(f"Enterprise app {r['name']} is not in the published list of apps; it is priced as one app to replace.",
                  f"Enterprise app {r['name']} is not in our published list of Enterprise apps and their replacements. "
                  f"This report prices it as one app to replace if you move to Community. We will confirm with you "
                  f"what it does and how it would be replaced, which can change the price of moving to Community.")
    apps.sort(key=lambda x: x["label"])

    enterprise_db = scan["odoo"]["web_enterprise_installed"] or bool(by_origin.get("enterprise"))
    cats = {}
    total = 0
    for k, label, meaning in CATEGORIES:
        items = scan["customizations"].get(k) or []
        cats[k] = {"label": label, "meaning": meaning, "count": len(items)}
        if k == "server_actions":
            cats[k]["runs_code"] = sum(1 for x in items if x.get("state") == "code")
        if k in ("custom_fields", "custom_views", "custom_models", "reports"):
            cats[k]["studio"] = sum(1 for x in items if x.get("studio"))
        if k == "automations":
            cats[k]["runs_code"] = sum(1 for x in items if x.get("runs_code"))
        total += len(items)

    companies = scan["companies"]["companies"]
    branches = scan["companies"]["branches"]
    users = scan["users"]["internal"]
    q_managed = managed({"companies": companies, "branches": branches, "users": users})
    q_rescue = rescue({"companies": companies, "databases": 1})
    q_cleanup = cleanup({"customizations": total})
    upgrade_block = None
    de_enterprise = None
    if major:
        if major < tgt:
            base_in = {"fromVersion": major, "toVersion": tgt, "customModules": to_port}
            upgrade_block = {"from": major, "to": tgt, "note": tgt_note,
                             "keep_enterprise": upgrade(dict(base_in, enterpriseApps=0)),
                             "to_community": upgrade(dict(base_in, enterpriseApps=len(apps))) if apps else None}
        elif apps:
            # already on the target: de-Enterprise alone, at the published price per app
            de_enterprise = upgrade({"fromVersion": major, "toVersion": major, "customModules": 0, "enterpriseApps": len(apps)})
    else:
        check("The Odoo version could not be read: the upgrade price needs a manual check.",
              "The scan could not read your Odoo version, so this report has no upgrade price yet and cannot say "
              "whether the database can leave Odoo Online. We will confirm the version with you. The upgrade price "
              "depends on the number of major versions between your version and the target.")
    leave = leave_online(v, enterprise_db, bool(upgrade_block and upgrade_block.get("to_community")) or bool(de_enterprise))

    # The scan's date, and the report's. scanned_at_utc is a field of the customer's own file, so it is checked.
    scan_time = parse_scan_time(scan.get("scanned_at_utc"))
    scan_day = scan_time[0] if scan_time else None
    listed = module_data_date(D)
    covers = ("The project fees cover what existed in your database on the scan date, so that date decides what "
              "they cover.")
    if not scan_day:
        check("The scan file's date (scanned_at_utc) is missing or is not a real date and time. Confirm with the "
              "customer when the scan was run; this report states no validity date.",
              "The scan file's date could not be read, so we will confirm with you when the scan was run before the "
              "quote is final. " + covers)
    elif scan_day < SCAN_TOOL_SINCE:
        check(f"The scan file is dated {scan_day.isoformat()}, before this scan tool existed. Confirm with the "
              f"customer when the scan was run; this report states no validity date from that date.",
              f"The scan file is dated {long_date(scan_day)}, before this scan tool was released, so we will confirm "
              f"with you when the scan was run before the quote is final. " + covers)
        scan_day = None
    elif issued and (scan_day - issued).days > SCAN_AFTER_ISSUE_DAYS:
        check(f"The scan file is dated {scan_day.isoformat()}, later than the report date {issued.isoformat()}. "
              f"Confirm with the customer when the scan was run.",
              f"The scan file is dated {long_date(scan_day)}, which is later than this report, so we will confirm "
              f"with you when the scan was run before the quote is final. " + covers)
        scan_day = None
    elif not issued and listed and (scan_day - listed).days > SCAN_AFTER_MODULE_LIST_DAYS:
        check(f"The scan file is dated {scan_day.isoformat()}, more than {SCAN_AFTER_MODULE_LIST_DAYS} days after the "
              f"module list ({listed.isoformat()}), and no --issued date was passed. Confirm with the customer when "
              f"the scan was run, or refresh the module list.",
              f"The scan file is dated {long_date(scan_day)}, much later than the module list this report uses, so we "
              f"will confirm with you when the scan was run before the quote is final. " + covers)
        scan_day = None
    report_day = issued or scan_day            # scan_day is None here when it is not believable
    try:
        valid_until = report_day + datetime.timedelta(days=QUOTE_VALID_DAYS) if report_day and not checks else None
    except OverflowError:                      # a date at the very end of the calendar: no last day is stated
        valid_until = None
    return {
        "scan": {k: scan.get(k) for k in ("database", "scanned_at_utc", "tool", "postgres", "filestore", "odoo", "companies", "users",
                                         "not_counted", "read_only_proof")},
        "module_data": D.get("generated") or D.get("source"),
        "target": tgt,
        "edition": "Enterprise" if enterprise_db else "Community",
        "version": v,
        "modules": {"installed": len(rows), "excluded": excluded, "pending_operations": pending,
                    "by_origin": {k: sorted(r["name"] for r in rs) for k, rs in sorted(by_origin.items())},
                    "to_port": to_port, "oca_missing_on_target": a["ocaMissingOnTarget"],
                    "community_not_on_target": (sorted(r["name"] for r in rows if r["cls"] == "community" and r["onTarget"] is False)
                                                if version_bit(D, tgt) else []),
                    "rows": rows, "overrides": eff_over},
        "enterprise_apps": apps,
        "customizations": {"categories": cats, "total": total},
        "quote": {"managed": q_managed, "rescue": q_rescue, "upgrade": upgrade_block, "de_enterprise": de_enterprise,
                  "cleanup": q_cleanup},
        "leave_online": leave,
        "checks": checks,
        "checks_customer": checks_customer,
        "preliminary": bool(checks),
        # The report's date: --issued, or the scan's date when none was passed; null when there is no usable date.
        "issued": report_day.isoformat() if report_day else None,
        # The last day the quote holds; null for a preliminary report (its 30 days start with the confirmed quote).
        "valid_until": valid_until.isoformat() if valid_until else None,
        "scan_date": scan_day.isoformat() if scan_day else None,      # the scan's date (UTC) when it is believable
        "report_wording": REPORT_WORDING,
    }


def leave_online(v, enterprise_db, priced_both):
    if not v:
        return {"can_leave": None, "answer": "The scan could not read your Odoo version, so we confirm the version "
                                             "with you before answering this."}
    if v["intermediate"]:
        nxt = v["major"] + 1
        return {"can_leave": False, "next_major": f"{nxt}.0",
                "answer": (f"Not yet. The database is on {v['raw']}, an intermediate version that only Odoo Online runs. "
                           f"Odoo has to upgrade it to {nxt}.0 (or a later major version) on Odoo Online first; you can "
                           f"ask for that upgrade from your database manager on odoo.com, or wait for the scheduled one. "
                           f"After that it can move to your own server.")}
    ans = f"Yes. The database is on Odoo {v['major']}.0, a major version, so it can run on your own server."
    if enterprise_db:
        ans += (" It uses Odoo Enterprise: on your own server you either keep an Odoo Enterprise subscription that "
                "covers self-hosting, or replace the Enterprise apps" + (" (the quote prices both)." if priced_both else "."))
    return {"can_leave": True, "answer": ans}


# ---------------------------------------------------------------------------------------------------------------------
ORIGIN_TEXT = {
    "community": ("Standard Odoo Community", "Part of Odoo itself; carried over at no charge."),
    "enterprise": ("Odoo Enterprise", "Odoo's paid modules. Apps are priced below; supporting modules come out with their app."),
    "oca": ("OCA", "Free modules from the Odoo Community Association; free to carry over when released for the target version."),
    "thirdparty": ("Third-party", "Written by a vendor or partner. Counted as modules to port unless the vendor supplies the new version."),
    "custom": ("Custom", "No author, or Odoo's default author: written for you. Counted as modules to port."),
    "vendor_supplied": ("Third-party, supplied by the vendor", "You confirmed the vendor supplies the new version. Not priced."),
    "dropped": ("Dropped on review", "You told us these are not carried over. Not priced."),
}
ORIGIN_ORDER = ["custom", "thirdparty", "oca", "enterprise", "community", "vendor_supplied", "dropped"]


def human_bytes(n):
    n = float(n or 0)
    for u in ("B", "KB", "MB", "GB", "TB"):
        if n < 1024 or u == "TB":
            return f"{n:.0f} {u}" if u == "B" else f"{n:.1f} {u}"
        n /= 1024


def report_lines(r):
    """The report as a list of blocks: ('h', level, text) | ('p', text) | ('table', header, rows) | ('ul', items).
    A text is a string, or a list of strings in which a Code is a technical name (<code> in HTML). Everything here is
    read by the customer. The dates come from build_report (r["issued"], r["scan_date"]): the clock is never read,
    so the same scan.json with the same arguments always gives the same report."""
    s = r["scan"]
    v = r["version"]
    prelim = bool(r.get("preliminary"))     # set by build_report when there is at least one point to confirm
    report_day = parse_date(r.get("issued"))
    scan_day = parse_date(r.get("scan_date"))
    valid_until = parse_date(r.get("valid_until"))
    scan_time = parse_scan_time(s.get("scanned_at_utc")) if scan_day else None
    at_clock = f" at {scan_time[1]}" if scan_time and scan_time[1] else ""
    tool = re.match(r"^allodi_scan\.py (\S+)$", str(s.get("tool") or ""))
    tool_txt = f"Allodi's scan tool, version {tool.group(1)}" if tool else str(s.get("tool") or "Allodi's scan tool")
    if report_day and scan_day == report_day:
        head = f"Scan and report date: {long_date(report_day)}. Scanned{at_clock} with {tool_txt}."
    elif report_day and scan_day:
        head = f"Report date: {long_date(report_day)}. Scanned on {long_date(scan_day)}{at_clock} with {tool_txt}."
    elif report_day:
        head = f"Report date: {long_date(report_day)}. Scanned with {tool_txt}; the date of the scan is to be confirmed."
    else:
        head = f"Scanned with {tool_txt}. The date of the scan is to be confirmed, so this report carries no date yet."
    listed = module_data_date({"generated": r.get("module_data")})
    B = []
    B.append(("h", 1, f"Allodi free scan: {s['database']}"))
    B.append(("p", head + " The scan read Odoo's settings tables only, in a read-only session, and changed nothing."
                   + (f" Module list dated {long_date(listed) if listed else r['module_data']}." if r.get("module_data") else "")))
    if report_day and scan_day and (report_day - scan_day).days > OLD_SCAN_DAYS:
        # Statement of work 11.3: the fees are binding for what existed on the Scan Date, not on the report's date.
        B.append(("p", f"The scan was run {(report_day - scan_day).days} days before this report; the fixed fees cover "
                       "what existed on the scan date."))
    B.append(("h", 2, "Summary"))
    vtxt = (f"Odoo {v['major']}" + (f" (intermediate version {v['raw']})" if v["intermediate"] else f" ({v['raw']})")) if v else "unknown"
    B.append(("table", ["What", "Count", "What it means"], [
        ["Odoo version", vtxt, "Read from the base module."],
        ["Edition", r["edition"], "Enterprise when any Odoo Enterprise module is installed."],
        ["Companies", str(s["companies"]["companies"]), "Active companies without a parent company."],
        ["Branches", str(s["companies"]["branches"]), "Active companies with a parent (Odoo 17 and later)."],
        ["Internal users", str(s["users"]["internal"]), "Active users with the Internal User type. Priced per user."],
        ["Portal users", str(s["users"]["portal"]), "Active portal users. Free."],
        ["Installed modules", str(r["modules"]["installed"]),
         (["All installed modules, by type below. Studio's own technical module (", Code("studio_customization"),
           ") is left out of this count: what you built with Studio is counted under customizations."]
          if "studio_customization" in r["modules"]["excluded"] else "All installed modules, by type below.")],
        ["Custom modules to port", str(r["modules"]["to_port"]), f"Custom and third-party modules, plus OCA modules without a release for Odoo {r['target']}."],
        ["Enterprise apps installed", str(len(r["enterprise_apps"])), "Priced only if you replace them with Community or OCA."],
        ["Customizations", str(r["customizations"]["total"]), "Everything made in the database rather than in module code, by category below."],
        ["Database size", human_bytes(s["postgres"]["database_bytes"]), "Size in PostgreSQL."],
        ["Files (filestore)", (f"{s['filestore']['files']} files, {human_bytes(s['filestore']['bytes'])}" if s.get("filestore") else "not measured"),
         "Attachments are stored outside the database; the scan never opens them."],
    ]))
    B.append(("h", 2, "Can you leave Odoo Online now?"))
    q = r["quote"]
    up = q["upgrade"]
    B.append(("p", r["leave_online"]["answer"] + " (This applies if the database is on Odoo Online today; the scan cannot "
                   "see where it is hosted.)"))
    if v and v["intermediate"] and up:
        # Mark, 2026-10-01: "17.2 is 17". The price rule is unchanged; this sentence states it next to the answer above.
        B.append(("p", f"The upgrade price below counts major versions from {up['from']}, the major line your database "
                       f"is on today, to Odoo {up['to']}: an intermediate version such as {v['raw']} is priced as "
                       f"Odoo {up['from']}."))

    # The quote. Contract basis (legal/02 statement of work 11.1, 11.3, 11.6, 11A; legal/03 service description 6.5,
    # 6.6, 11.2a): only the fixed project fees are binding; the monthly fee follows the daily counts.
    B.append(("h", 2, "Your quote (preliminary)" if prelim else "Your quote"))
    if prelim:
        B.append(("p", "These figures are preliminary. The points listed under \"Before this quote is final\" are "
                       "confirmed with you first, and some of them can change a price. The confirmed quote, with "
                       "binding project fees, follows once they are confirmed."))
    # A preliminary report promises no 30 days from its own date (the date may be the very thing to confirm): the 30
    # days start with the confirmed quote. A report that is not preliminary always has a date (build_report).
    B.append(("p", "Every number below is a count above multiplied by the published prices at allodi.us/pricing.html"
                   + (f" on the date of this report, {long_date(report_day)}. " if report_day else ". ")
                   + (f"The quote holds for {QUOTE_VALID_DAYS} days from that date, through {long_date(valid_until)}. "
                      if valid_until and not prelim else
                      f"The quote, once confirmed, holds for {QUOTE_VALID_DAYS} days from the date we send the "
                      "confirmed quote. ")
                   + "All amounts are in US dollars and exclude sales tax, which is added where it applies."))
    m = q["managed"]
    rows = [[l["label"], str(l["qty"]), money(l["amount"])] for l in m["lines"] if l["qty"] or l["label"] == "First company"]
    rows.append(["Monthly total", "", money(m["monthly"])])
    B.append(("h", 3, "Managed Odoo, monthly"))
    B.append(("table", ["Line", "Count", "Per month"], rows))
    B.append(("p", "The monthly total is an estimate from the counts on the scan date. Each month's fee follows the "
                   "daily counts of your companies, branches and internal users: we count them every day, and each "
                   "invoice uses the highest daily count of the billing period that just ended, as the Managed Odoo "
                   "service description provides."))
    B.append(("p", f"The fee includes {m['includedHours']:g} fix hours a month at these counts. Time beyond the included "
                   f"hours is billed at {money(m['hourly'])} an hour in 15-minute increments; we warn you before going "
                   "over and do not exceed the included time by more than 2 hours without your approval. Work "
                   "estimated at more than 8 hours is quoted separately."))
    B.append(("p", "Hosting is billed separately. In the standard setup, the server and backups are in your own Hetzner "
                   "and Backblaze accounts and are billed to you by them. Hosting we provide temporarily during a move "
                   "is passed through at cost."))
    B.append(("h", 3, "One-time project fees"))
    counted = "modules, Enterprise apps, companies, branches, internal users and customizations"
    if prelim:
        B.append(("p", "The project fees below (rescue and rehost, upgrade, customization cleanup) are worked out from "
                       f"what the scan counts: {counted}. They are preliminary until the points under \"Before this "
                       "quote is final\" are confirmed with you."))
    else:
        B.append(("p", "The project fees below (rescue and rehost, upgrade, customization cleanup) are fixed, and they "
                       f"are binding on us for what the scan counts: {counted}. If one of those existed in your "
                       "database on the scan date and the scan did not count it, we do the work on it at our own cost "
                       "and the fees stay the same. Items added after the scan date are handled as a change under the "
                       "statement of work: priced from the published prices where they cover it, otherwise by the hour "
                       "with an estimate first. What the scan does not count, such as integrations (listed at the "
                       "end), is agreed with you before work starts."))
    rs = q["rescue"]
    B.append(("h", 3, "Rescue and rehost, one time"))
    B.append(("table", ["Milestone", "Amount"], [[x["label"], money(x["amount"])] for x in rs["milestones"]] + [["Total", money(rs["total"])]]))
    B.append(("p", "The first milestone is invoiced when the statement of work is signed; the server build and the "
                   "restored test copy are the work that follows. If we cannot get your database running on the new "
                   "server, with all modules loading and the data integrity checks passing, we refund the milestone "
                   "paid at signing in full, unless the failure is caused by a dependency on your side that is not "
                   "met, by you withdrawing our access, or by a third party refusing to release your data when you "
                   "ask them to."))
    B.append(("h", 3, "Upgrade, one time"))
    apps_txt = plural(len(r["enterprise_apps"]), "Enterprise app")
    if not up and not v:
        B.append(("p", "The scan could not read your Odoo version, so the upgrade is not priced here. We confirm the "
                       "version with you first (see \"Before this quote is final\")."))
    elif not up:
        B.append(("p", f"Not needed: the database is already on Odoo {v['major']}, at or past the target Odoo {r['target']}."))
        de = q.get("de_enterprise")
        if de:
            B.append(("p", f"Moving from Enterprise to Community on this version: {apps_txt} "
                           f"replaced x {money(PRICES['upgrade']['perEnterpriseApp'])} = {money(de['total'])}. Keeping Enterprise: $0."))
    else:
        if up.get("note"):
            B.append(("p", up["note"]))
        k = up["keep_enterprise"]
        B.append(("table", ["Line", "Count", "Amount"], [[l["label"], str(l["qty"]), money(l["amount"])] for l in k["lines"][:2]] +
                  [[f"Upgrade Odoo {up['from']} to {up['to']}" + (", keeping Enterprise" if up.get("to_community") else ""), "", money(k["total"])]]))
        if up.get("to_community"):
            c = up["to_community"]
            B.append(("p", f"Moving to Community instead adds {apps_txt} replaced x "
                           f"{money(PRICES['upgrade']['perEnterpriseApp'])} = {money(c['lines'][2]['amount'])}, for a total of {money(c['total'])}."))
    cl = q["cleanup"]
    B.append(("h", 3, "Customization cleanup, one time"))
    B.append(("p", f"{cl['items']} customizations: {money(cl['total'])}." if cl["items"] else "No customizations found: nothing to clean up."))

    if prelim:
        # Only the customer wording is printed. r["checks"] is the internal copy for Allodi's staff (the lead's note).
        B.append(("h", 2, "Before this quote is final"))
        B.append(("p", "The figures in this report are preliminary until we have confirmed the points below with you. "
                       "Each point says what we will confirm and why it matters for the price. Once they are "
                       "confirmed, we send you the confirmed quote, with binding project fees."))
        names = {x["name"] for x in r["modules"]["rows"]} | {x["name"] for x in r["enterprise_apps"]}
        items = []
        for text in r["checks_customer"]:
            m = _CHECK_NAME.match(text)        # a point about one module: its technical name is marked as a name
            if m and m.group(2) in names:
                items.append([m.group(1) + " ", Code(m.group(2)), text[m.end(2):]])
            else:
                items.append(text)
        B.append(("ul", items))

    B.append(("h", 2, "Modules by type"))
    trows = []
    for key in ORIGIN_ORDER:
        names = r["modules"]["by_origin"].get(key)
        if names:
            t, meaning = ORIGIN_TEXT[key]
            trows.append([t, str(len(names)), meaning])
    B.append(("table", ["Type", "Installed", "What it means"], trows))
    port_rows = [x for x in r["modules"]["rows"] if x["ports"]]
    if port_rows:
        B.append(("h", 3, "Modules counted to port"))
        B.append(("table", ["Module", "Type", "Why"], [[Code(x["name"]), ORIGIN_TEXT.get(x["origin"], (x["origin"],))[0],
                                                       x["why"] or ("OCA, no release for Odoo %s yet" % r["target"] if x["cls"] == "oca" else "")] for x in port_rows]))
        B.append(("p", "If one of these is wrong (a vendor sells the new version and you buy it, or you will not keep the module), "
                       "tell us and we re-issue the quote without it."))
    if r["enterprise_apps"]:
        B.append(("h", 3, "Enterprise apps installed"))
        B.append(("table", ["App", "Community or OCA replacement"],
                  [[Code(x["label"]) if x.get("unlisted") else x["label"], x.get("note") or ""] for x in r["enterprise_apps"]]))
    if r["modules"]["community_not_on_target"]:
        B.append(("p", ["Standard modules that do not exist in Odoo %s (merged or retired by Odoo; handled in the upgrade "
                        "at no charge): " % r["target"]] + name_list(r["modules"]["community_not_on_target"]) + ["."]))
    if r["modules"]["pending_operations"]:
        B.append(("p", ["Modules waiting for an install or upgrade that never finished (not counted): "]
                       + name_list(r["modules"]["pending_operations"]) + ["."]))

    B.append(("h", 2, "Customizations by category"))
    crow = []
    for k, label, meaning in CATEGORIES:
        c = r["customizations"]["categories"][k]
        extra = []
        if c.get("studio"):
            extra.append(f"{c['studio']} made with Studio")
        if c.get("runs_code"):
            extra.append(f"{c['runs_code']} run Python code")
        crow.append([label, str(c["count"]), meaning + (" " + "; ".join(extra) + "." if extra else "")])
    crow.append(["Total", str(r["customizations"]["total"]), "The number the cleanup price uses."])
    B.append(("table", ["Category", "Count", "What it means"], crow))

    B.append(("h", 2, "What the scan did not count"))
    nc = s["not_counted"]
    B.append(("ul", [
        f"Website content: {nc['website_views']} website pages and per-website copies of views made with the website builder. They are content, not customizations.",
        f"Menus ({nc['menus_made_in_interface']}) and window actions ({nc['window_actions_made_in_interface']}) made in the interface. They come with the model, report or action they open.",
        f"Analytic plan columns ({nc.get('analytic_plan_columns', 0)}): fields named x_planN_id that Odoo itself adds for each extra analytic plan.",
        f"Views Odoo rewrites by itself ({nc.get('views_odoo_rewrites', 0)}), such as the access-rights part of the user form: marked as changed, but not by anyone.",
        "Your records, email templates, filters, translations and system parameters, and who your users are. The scan counts users but never reads their names, logins or passwords.",
        "Code in custom modules: the scan counts modules, it does not read their source. The upgrade price is per module.",
        "Custom property fields (Odoo 16 and later), which are stored with your records, and anything in the filestore.",
        "Integrations with outside systems: listed with you in the statement of work, not by the scan.",
        "Other databases on the same server: each database is scanned and priced on its own.",
    ]))
    B.append(("p", "The detail appendix lists every module and customization by name. It holds names only, never your records."))
    return B


def appendix_blocks(r, scan):
    scan_day = parse_date(r.get("scan_date"))
    B = [("h", 1, f"Appendix: names found by the scan ({scan['database']}"
                  + (f", scanned {long_date(scan_day)})" if scan_day else ")")),
         ("p", "Names only. The scan never reads records, code bodies, rule domains, logins or company names.")]
    B.append(("h", 2, "Installed modules"))
    B.append(("table", ["Module", "Type", "Version", "Author", "License", "Counted to port"],
              [[Code(x["name"]), ORIGIN_TEXT.get(x["origin"], (x["origin"],))[0], x.get("version") or "", x.get("author") or "",
                x.get("license") or "", "yes" if x["ports"] else ""] for x in r["modules"]["rows"]]))
    cz = scan["customizations"]
    fmt = {
        "custom_models": (["Model", "Label", "Studio", "Counted with it (fields / views / access / rules)"],
                          lambda x: [x["model"], x.get("label") or "", "yes" if x.get("studio") else "",
                                     ("{fields} / {views} / {access_rights} / {record_rules}".format(**x["scaffolding"]) if x.get("scaffolding") else "")]),
        "custom_fields": (["Model", "Field", "Type", "Label", "Studio"], lambda x: [x["model"], x["name"], x["ttype"], x.get("label") or "", "yes" if x.get("studio") else ""]),
        "custom_views": (["Model", "View", "Type", "Inherits", "Active", "Studio"], lambda x: [x.get("model") or "", x.get("name") or "", x.get("type") or "", "yes" if x.get("inherited") else "", "yes" if x.get("active") else "no", "yes" if x.get("studio") else ""]),
        "edited_views": (["Model", "View", "Type", "Key", "Active"], lambda x: [x.get("model") or "", x.get("name") or "", x.get("type") or "", x.get("key") or "", "yes" if x.get("active") else "no"]),
        "server_actions": (["Model", "Action", "Kind", "Code length"], lambda x: [x.get("model") or "", x.get("name") or "", x.get("state") or "", str(x.get("code_chars") or "")]),
        "automations": (["Model", "Rule", "Trigger", "Actions", "Active"], lambda x: [x.get("model") or "", x.get("name") or "", x.get("trigger") or "", x.get("action_types") or "", "yes" if x.get("active") else "no"]),
        "scheduled_jobs": (["Job", "Model", "Every", "Active"], lambda x: [x.get("name") or "", x.get("model") or "", f"{x.get('interval_number')} {x.get('interval_type')}", "yes" if x.get("active") else "no"]),
        "reports": (["Model", "Report", "Template", "Type"], lambda x: [x.get("model") or "", x.get("name") or "", x.get("report_name") or "", x.get("report_type") or ""]),
        "access_rights": (["Model", "Name", "Active"], lambda x: [x.get("model") or "", x.get("name") or "", "yes" if x.get("active") else "no"]),
        "record_rules": (["Model", "Name", "Active"], lambda x: [x.get("model") or "", x.get("name") or "", "yes" if x.get("active") else "no"]),
        "approval_rules": (["Model", "Rule", "Active"], lambda x: [x.get("model") or "", x.get("name") or "", "yes" if x.get("active") else "no"]),
        "assets": (["Name", "Bundle", "Path", "Active"], lambda x: [x.get("name") or "", x.get("bundle") or "", x.get("path") or "", "yes" if x.get("active") else "no"]),
    }
    for k, label, _m in CATEGORIES:
        items = cz.get(k) or []
        B.append(("h", 2, f"{label} ({len(items)})"))
        if items:
            head, f = fmt[k]
            B.append(("table", head, [f(x) for x in items]))
        else:
            B.append(("p", "None."))
    return B


class Code(str):
    """A technical name (a module's) inside the report's text: <code>...</code> in HTML, so a page can let just the
    names break on a narrow screen. Markdown prints it like any other text. Escaped like everything else."""


_CHECK_NAME = re.compile(r"^(Module|Enterprise app) (.+?) (?=states Odoo's Enterprise license|is not in our published list)")


def name_list(names):
    """['a', 'b'] -> [Code('a'), ', ', Code('b')]: names in a sentence."""
    out = []
    for n in names:
        if out:
            out.append(", ")
        out.append(Code(n))
    return out


def plain_text(x):
    """A text of the report (a string, or a list of strings and Codes) as one string."""
    return x if isinstance(x, str) else "".join(str(p) for p in x)


def html_text(x):
    """A text of the report as HTML: every part escaped, a Code inside <code>."""
    return "".join(f"<code>{html.escape(p)}</code>" if isinstance(p, Code) else html.escape(str(p))
                   for p in ([x] if isinstance(x, str) else x))


def md_escape(t):
    t = plain_text(t).replace("\r", " ").replace("\n", " ")
    return re.sub(r"([\\`*_{}\[\]<>|#])", r"\\\1", t)


def to_markdown(blocks):
    out = []
    for b in blocks:
        if b[0] == "h":
            out.append("#" * b[1] + " " + md_escape(b[2]))
        elif b[0] == "p":
            out.append(md_escape(b[1]))
        elif b[0] == "ul":
            out.append("\n".join("- " + md_escape(x) for x in b[1]))
        elif b[0] == "table":
            out.append("| " + " | ".join(md_escape(h) for h in b[1]) + " |\n|" + "---|" * len(b[1]) + "\n" +
                       "\n".join("| " + " | ".join(md_escape(c) for c in row) + " |" for row in b[2]))
        out.append("")
    return "\n".join(out)


HTML_HEAD = """<!DOCTYPE html>
<html lang="en"><head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1">
<meta name="robots" content="noindex"><title>{title}</title>
<style>
:root{{--bg:#fff;--fg:#1d2320;--muted:#5b6660;--line:#d9dfdb;--accent:#1f6f4a;--tint:#f3f7f4}}
@media (prefers-color-scheme:dark){{:root{{--bg:#131715;--fg:#e7ece9;--muted:#a3ada7;--line:#2e3632;--accent:#6fcf9f;--tint:#1a201d}}}}
body{{background:var(--bg);color:var(--fg);font:16px/1.55 system-ui,-apple-system,"Segoe UI",sans-serif;margin:0;padding:24px 16px}}
main{{max-width:880px;margin:0 auto}} h1{{font-size:1.6rem;margin:0 0 .5rem}} h2{{font-size:1.25rem;margin:2rem 0 .5rem;color:var(--accent)}}
h3{{font-size:1.05rem;margin:1.25rem 0 .4rem}} p,li{{max-width:70ch}} .wrap{{overflow-x:auto}}
table{{border-collapse:separate;border-spacing:0;width:100%;margin:.5rem 0 1rem;font-size:.95rem}}
th,td{{text-align:left;padding:.4rem .6rem;border-bottom:1px solid var(--line);vertical-align:top}}
thead th{{background:var(--tint)}} td:nth-child(2){{white-space:nowrap}}
</style></head><body><main>
"""


def to_html(blocks, title):
    e = html.escape
    out = [HTML_HEAD.format(title=e(title))]
    for b in blocks:
        if b[0] == "h":
            out.append(f"<h{b[1]}>{html_text(b[2])}</h{b[1]}>")
        elif b[0] == "p":
            out.append(f"<p>{html_text(b[1])}</p>")
        elif b[0] == "ul":
            out.append("<ul>" + "".join(f"<li>{html_text(x)}</li>" for x in b[1]) + "</ul>")
        elif b[0] == "table":
            out.append('<div class="wrap"><table><thead><tr>' + "".join(f'<th scope="col">{e(h)}</th>' for h in b[1]) +
                       "</tr></thead><tbody>" + "".join("<tr>" + "".join(f"<td>{html_text(c)}</td>" for c in row) + "</tr>" for row in b[2]) +
                       "</tbody></table></div>")
    out.append("</main></body></html>\n")
    return "\n".join(out)


def print_summary(scan, out=sys.stdout):
    o = scan["odoo"]
    w = lambda s="": print(s, file=out)  # noqa: E731
    w(f"Allodi scan of database '{_printable(scan['database'])}' at {scan['scanned_at_utc']} ({scan['tool']})")
    w(f"  read-only session: transaction_read_only={scan['read_only_proof']['transaction_read_only']} "
      f"via {scan['read_only_proof']['backend']} as {scan['read_only_proof']['db_user']}")
    w(f"  Odoo version: {o['base_version']}" + ("  (intermediate: Odoo Online only)" if o["intermediate"] else ""))
    w(f"  companies {scan['companies']['companies']}, branches {scan['companies']['branches']}, "
      f"internal users {scan['users']['internal']}, portal users {scan['users']['portal']}")
    inst = [m for m in scan["modules"] if m["state"] in INSTALLED_STATES]
    w(f"  installed modules: {len(inst)}; database size {human_bytes(scan['postgres']['database_bytes'])}")
    w("  customizations (names are in the file):")
    for k, label, _m in CATEGORIES:
        w(f"    {label:<28} {len(scan['customizations'].get(k) or [])}")
    w("  not counted: " + ", ".join(f"{k.replace('_', ' ')} {v}" for k, v in scan["not_counted"].items()))


# =====================================================================================================================
# CLI
# =====================================================================================================================
def parse_overrides(items):
    out = {}
    for it in items or []:
        m = re.match(r"^([A-Za-z0-9_]+)=(custom|thirdparty|enterprise|drop)$", it)
        if not m:
            raise SystemExit(f"--override must look like module_name=custom|thirdparty|enterprise|drop, not {it!r}")
        out[m.group(1)] = m.group(2)
    return out


def default_module_data():
    here = os.path.dirname(os.path.abspath(__file__))
    for cand in (os.path.join(here, "odoo-modules.json"), os.path.join(here, "..", "customer", "verify", "odoo-modules.json"),
                 os.path.join(here, "odoo-modules.js")):
        if os.path.isfile(cand):
            return cand
    return None


def write_reports(scan, a, outdir):
    path = a.module_data or default_module_data()
    if not path:
        raise ScanError("module list not found: pass --module-data odoo-modules.json (or odoo-modules.js from allodi.us/assets/)")
    D = load_module_data(path)
    r = build_report(scan, D, a.target, parse_overrides(a.override), getattr(a, "issued_date", None))
    os.makedirs(outdir, exist_ok=True)
    blocks = report_lines(r)
    app = appendix_blocks(r, scan)
    title = f"Allodi scan: {scan['database']}"
    files = {"report.json": json.dumps(r, indent=2, sort_keys=True, default=str),
             "report.md": to_markdown(blocks), "report.html": to_html(blocks, title),
             "appendix.md": to_markdown(app), "appendix.html": to_html(app, title + " (appendix)")}
    for n, content in files.items():
        with open(os.path.join(outdir, n), "w", encoding="utf-8", newline="\n") as f:
            f.write(content)
    return r


# ---------------------------------------------------------------------------------------------------------------------
# No --db and no PGDATABASE: read the list of database names, nothing else. One read-only session on the maintenance
# database; no other database is opened to look. The names go to the screen (stderr) only, never to a file, and only
# the chosen name is handed on to the scan. Odoo's configuration file is never opened.
# ---------------------------------------------------------------------------------------------------------------------
MAINTENANCE_DATABASES = ("postgres", "template1")   # every PostgreSQL server has them; the list is read from one
LOOKUP_CONNECT_TIMEOUT = 10                         # seconds, the list only: an unreachable server fails fast
LIST_AT_MOST = 50                                   # a longer list is counted, not printed
WHICH_DATABASE = "Which database? Pass --db NAME (the name shown in Odoo's database selector or in odoo.conf db_name)."
# Every database that allows connections, whether or not this login may open it: one it cannot open is still counted.
Q_DATABASES = ("SELECT d.datname AS name FROM pg_catalog.pg_database d WHERE NOT d.datistemplate AND d.datallowconn "
               "ORDER BY d.datname")
PLAIN_NAME = re.compile(r"^[A-Za-z0-9_.-]+\Z")      # a name that pastes into any shell as it is


def _needs_password(error):
    """libpq's own words for 'a password is needed' (none given) or 'the password given was refused'."""
    text = str(error).lower()
    return "no password supplied" in text or "password authentication failed" in text


NETWORK_WORDS = ("could not connect", "connection refused", "timeout expired", "could not translate host name",
                 "is the server running", "network is unreachable", "no route to host", "server closed the connection")


def _network_failure(error):
    """libpq's own words for 'the server could not be reached' (naming a database would not help)."""
    text = str(error).lower()
    return any(w in text for w in NETWORK_WORDS)


def _printable(text):
    """Text that came from the server, safe to show on a terminal: characters that do not print become '?'."""
    return "".join(ch if ch.isprintable() else "?" for ch in str(text))


def _reason(error):
    """The first line of an error, short enough for one line on the screen."""
    text = str(error)
    for prefix in ("psql failed: ", "cannot connect: ", "psql: error: ", "psql: "):
        if text.startswith(prefix):
            text = text[len(prefix):]
    lines = [" ".join(l.split()) for l in text.splitlines() if l.strip()]
    return _printable(lines[0] if lines else "no answer")[:300]


def _shell_word(word):
    """One word of a command the user can paste back into the shell they ran this from, or None if unsure."""
    word = str(word)
    if not word.isprintable():
        return None
    if os.name != "nt":
        return shlex.quote(word)
    if re.match(r"^[A-Za-z0-9_.:/\\=@,+-]+\Z", word):
        return word
    return '"' + word + '"' if not re.search(r"[\"%!$`]", word) else None   # cmd.exe and PowerShell


def _program():
    """How this run was started, for the commands shown to the user: 'python3 allodi_scan.py'."""
    exe = sys.executable or "python3"
    short = os.path.basename(exe)
    if short.lower().endswith(".exe"):
        short = short[:-4]
    found = shutil.which(short)
    if found and os.path.normcase(os.path.abspath(found)) == os.path.normcase(os.path.abspath(exe)):
        exe = short            # the plain command starts this same Python; otherwise show the full path
    script = "allodi_scan.py"
    if __name__ == "__main__":
        script = sys.argv[0] if sys.argv else ""
        if script in ("", "-", "-c") or not os.path.isfile(script):
            return None        # read from standard input (curl ... | python3 -): there is no file to name
    words = [_shell_word(exe), _shell_word(script)]
    return None if None in words else " ".join(words)


def _sudo_prefix():
    """'sudo -u odoo ' when this run was started with sudo as another user than root; None when it cannot tell."""
    if not os.environ.get("SUDO_USER") or not hasattr(os, "geteuid") or os.geteuid() == 0:
        return None
    started = os.environ.get("SUDO_COMMAND", "").split()
    if not started or not os.path.basename(started[0]).lower().startswith("python"):
        return None            # sudo started a shell or su (sudo su odoo, sudo -i): this user may have no sudo at all
    try:
        import pwd
        name = pwd.getpwuid(os.geteuid()).pw_name
    except (ImportError, KeyError):
        return None
    return f"sudo -u {name} " if PLAIN_NAME.match(name) else None


def _psql_list_command(a):
    """'psql -l' with the connection settings of this run, or None when there is no psql to run."""
    if not shutil.which(a.psql or "psql"):
        return None
    cmd = [a.psql or "psql", "-l"]
    for flag, val in (("-h", a.host), ("-p", a.port), ("-U", a.user)):
        if val:
            cmd += [flag, str(val)]
    words = [_shell_word(c) for c in cmd]
    return None if None in words else " ".join(words)


def _list_databases(a):
    """The names in pg_database, read in one read-only session on postgres (template1 if postgres is refused),
    opened exactly like the scan's own sessions, plus a connect timeout. Returns (maintenance database, names).
    The password is asked for at most once; it is kept for the scan only when PostgreSQL accepted it."""
    b = argparse.Namespace(**vars(a))
    b.ask_password, b.connect_timeout = False, LOOKUP_CONNECT_TIMEOUT
    typed, problem = None, None
    for maintenance in MAINTENANCE_DATABASES:
        b.db = maintenance
        while True:
            try:
                names = [str(r["name"]) for r in ReadOnlyDB(b).run([("q", Q_DATABASES)], check=False)["q"]]
                if typed is not None:
                    a.password = typed          # accepted: the scan of the chosen database may use it
                return maintenance, names
            except ScanError as e:
                problem = e
                if typed is None and _needs_password(e) and sys.stdin.isatty():
                    import getpass
                    try:
                        typed = b.password = getpass.getpass("PostgreSQL password: ")
                    except EOFError:
                        raise e
                    continue
                break
        if f'database "{maintenance}"' not in str(problem):
            break       # only a refusal of postgres itself is worth trying template1 for (not a password, role, host)
    raise problem


def find_database(a, argv):
    """Called when neither --db nor PGDATABASE is set. Reads the list of database names and either sets a.db (one
    database: returns None) or prints what to do and returns the exit status. It never picks between several."""
    def say(line=""):
        print(line, file=sys.stderr)

    def guidance(first, status, which=True):
        say(first)
        if which:
            say(WHICH_DATABASE)
        psql_l = _psql_list_command(a)
        if psql_l:
            say(f"To see the database names on this server, run: {psql_l}")
        return status

    try:
        _maintenance, names = _list_databases(a)
    except (ScanError, OSError) as e:
        # naming the database helps when only the list was refused, not when the server or the password was
        return guidance(f"The list of databases on this server could not be read. Reason: {_reason(e)}", 1,
                        which=not (_needs_password(e) or _network_failure(e)))
    except Exception as e:  # noqa: BLE001  an odd answer from psql or psycopg2: one plain line, no traceback
        return guidance("The list of databases on this server could not be read. "
                        f"Reason: {_reason(f'unexpected answer ({type(e).__name__}: {e})')}", 1)
    candidates = [n for n in names if n != "postgres"] or names
    if not candidates:
        return guidance("This PostgreSQL server has no database to scan. If Odoo uses another server, "
                        "pass --host and --port.", 1)
    if len(candidates) == 1:
        a.db = candidates[0]
        besides = " apart from 'postgres'" if len(names) > 1 else ""
        host = a.host or os.environ.get("PGHOST")
        port = _printable(a.port or os.environ.get("PGPORT") or "5432")
        where = (f"the PostgreSQL server at {_printable(host)}, port {port}" if host
                 else f"this computer's PostgreSQL server, port {port}")
        say(f"No --db was given. Using '{_printable(a.db)}', the only database{besides} on {where}. If Odoo uses "
            f"another server or port, add --host or --port. To scan a different one, add --db NAME.")
        return None
    say(f"No --db was given, and this PostgreSQL server has {len(candidates)} databases.")
    if len(candidates) > LIST_AT_MOST:
        return guidance("That is too many to list here.", 2)
    program = _program()
    rest = [_shell_word(x) for x in argv]
    sudo = _sudo_prefix()
    pasteable = program is not None and None not in rest
    say("Run the command for the one you want to scan:" if pasteable else "Add --db=NAME to your command.")
    for name in candidates:
        say()
        say(f"  {_printable(name)}")
        if not name.isprintable():
            psql_l = _psql_list_command(a)
            say("      This name has characters that cannot be shown here."
                + (f" To see it, run: {psql_l}" if psql_l else ""))
        elif not PLAIN_NAME.match(name):
            say("      To scan it, add --db and this name, in quotes as your shell needs.")
        elif not pasteable:
            say(f"      --db={name}")
        else:
            say("      " + (sudo or "") + " ".join([program] + rest + [f"--db={name}"]))
    from_env = [k for k in ("PGHOST", "PGPORT", "PGUSER") if os.environ.get(k)]
    if from_env:
        say()
        say(f"These commands rely on {' and '.join(from_env)} being set in your environment as they are now.")
    if pasteable and sudo is None:
        say()
        say("Run it the same way you ran this command, as the same user.")
    return 2


def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("command", nargs="?", default="collect", choices=["collect", "report", "scan", "selftest-readonly"])
    ap.add_argument("--db", default=os.environ.get("PGDATABASE"),
                    help="database name (default $PGDATABASE; with neither, the scan reads the list of database "
                         "names on the server and nothing else: one database is used, several are listed)")
    ap.add_argument("--host")
    ap.add_argument("--port")
    ap.add_argument("--user")
    ap.add_argument("--psql", help="path to psql (default: psql on PATH)")
    ap.add_argument("--backend", default="auto", choices=["auto", "psycopg2", "psql"])
    ap.add_argument("--timeout", type=int, default=120, help="per-statement timeout in seconds (default 120)")
    ap.add_argument("--filestore", help="optional: the database's filestore folder, to report its size (file names and sizes only)")
    ap.add_argument("--out", help="collect: output JSON file; report/scan: output folder")
    ap.add_argument("--scan-json", help="report: the scan.json to report on")
    ap.add_argument("--module-data", help="report: odoo-modules.json or odoo-modules.js")
    ap.add_argument("--target", type=int, default=DEFAULT_TARGET, help="upgrade target major version (default 19)")
    ap.add_argument("--override", action="append", help="report: module=custom|thirdparty|enterprise|drop after review")
    ap.add_argument("--issued", metavar="YYYY-MM-DD",
                    help="report: the date the report is issued; its prices and the quote's 30 days are counted from "
                         "it (default: the date of the scan). The report never reads the clock.")
    ap.add_argument("--quiet", action="store_true")
    ap.add_argument("--allow-superuser", action="store_true",
                    help="run even when this login is a PostgreSQL superuser and Odoo's own database user is not "
                         "(not recommended)")
    a = ap.parse_args(argv)
    a.issued_date = None
    if a.issued is not None:
        a.issued_date = parse_date(a.issued)
        # refused before anything is read or written; a report cannot be older than this tool
        if a.issued_date is None or not SCAN_TOOL_SINCE <= a.issued_date <= ISSUED_LATEST:
            print(f"--issued must be a real date written YYYY-MM-DD (for example 2026-10-01), from "
                  f"{SCAN_TOOL_SINCE.isoformat()} on, not {a.issued!r}", file=sys.stderr)
            return 2
    picked = False                                  # the database was found by reading the list (no --db)
    try:
        if a.command in ("collect", "scan", "selftest-readonly") and a.db is None:
            status = find_database(a, sys.argv[1:] if argv is None else list(argv))
            if status is not None:
                return status
            picked = True
        if a.command in ("collect", "scan", "selftest-readonly") and not a.db:     # --db '' or PGDATABASE=''
            print(WHICH_DATABASE, file=sys.stderr)
            return 2
        if a.command == "selftest-readonly":
            if picked:
                print(f"Self-test on database '{_printable(a.db)}' (no --db was given, so the scan picked it).")
            return selftest_readonly(a)
        if a.command in ("collect", "scan"):
            scan = collect(a)
            if a.command == "collect":
                out = a.out or f"allodi-scan-{re.sub(r'[^A-Za-z0-9_.-]', '_', scan['database'])}-{scan['scanned_at_utc'][:10]}.json"
            else:
                outdir = a.out or f"allodi-scan-{re.sub(r'[^A-Za-z0-9_.-]', '_', scan['database'])}"
                os.makedirs(outdir, exist_ok=True)
                out = os.path.join(outdir, "scan.json")
            with open(out, "w", encoding="utf-8", newline="\n") as f:
                json.dump(scan, f, indent=2, sort_keys=True, default=str)
            if not a.quiet:
                print_summary(scan)
                print(f"\nWrote {out}. It holds counts and names only. Read it, then send it to Allodi.")
            if a.command == "scan":
                r = write_reports(scan, a, outdir)
                if not a.quiet:
                    print(f"Wrote report.md, report.html, appendix.md, appendix.html in {outdir}: "
                          f"{r['modules']['to_port']} modules to port, {len(r['enterprise_apps'])} Enterprise apps, "
                          f"{r['customizations']['total']} customizations.")
            return 0
        if a.command == "report":
            if not a.scan_json:
                print("report needs --scan-json FILE", file=sys.stderr)
                return 2
            with open(a.scan_json, encoding="utf-8") as f:
                scan = json.load(f)
            if scan.get("format") != SCAN_FORMAT:
                raise ScanError(f"{a.scan_json} is not a scan.json of format {SCAN_FORMAT}")
            outdir = a.out or os.path.dirname(os.path.abspath(a.scan_json))
            r = write_reports(scan, a, outdir)
            if not a.quiet:
                print(f"Wrote reports in {outdir}: {r['modules']['to_port']} modules to port, "
                      f"{len(r['enterprise_apps'])} Enterprise apps, {r['customizations']['total']} customizations, "
                      f"cleanup {money(r['quote']['cleanup']['total'])}, managed {money(r['quote']['managed']['monthly'])}/month.")
            return 0
    except ScanError as e:
        print("ERROR: " + "\n".join(_printable(line) for line in str(e).split("\n")), file=sys.stderr)
        if picked and "does not look like an Odoo database" in str(e):
            print("It was the only database found there. If Odoo uses another server or port, add --host or --port.",
                  file=sys.stderr)
        return 1
    except OSError as e:
        print(f"ERROR: {_printable(e.strerror or e)} ({_printable(e.filename or '')}). Run it from a folder this user "
              f"can write to (for example: cd /tmp), or pass --out with a writable path.", file=sys.stderr)
        return 1
    return 2


def selftest_readonly(a):
    """Proves the session cannot write: tries an UPDATE that would change nothing even if it ran, inside the same
    read-only session the scan uses, and expects PostgreSQL to refuse it (SQLSTATE 25006)."""
    db = ReadOnlyDB(a)
    msg = db.try_write("UPDATE public.ir_module_module SET name = name WHERE false")
    if msg is None:
        print("FAIL: a write statement was accepted")
        return 1
    if "read-only transaction" in msg:
        print(f"PASS: PostgreSQL refused a write in the scan's session ({msg.strip().splitlines()[0]})")
        return 0
    print(f"FAIL: unexpected error: {msg}")
    return 1


if __name__ == "__main__":
    sys.exit(main())