# -*- coding: utf-8 -*- import logging import re import secrets import threading import odoo from odoo import SUPERUSER_ID, api, fields, models from odoo.exceptions import UserError from odoo.service import db as db_service _logger = logging.getLogger(__name__) # Readable temp-password parts (operator hands them over; admin changes them). _PW_WORDS = ["mardi", "olive", "safran", "tunis", "carthage", "jasmin", "menthe", "sfax", "gabes", "ariana", "sousse", "medina"] class NacefTenant(models.Model): """A taxpayer (contribuable) managed by the SaaS operator. In the DB-per-taxpayer model each tenant maps to one Odoo database (``db_name``). The control plane tracks the tenant lifecycle and metadata; it never stores the tenant's fiscal/encaissement data. """ _name = "nacef.tenant" _description = "NACEF Tenant (Taxpayer)" _order = "name" name = fields.Char(required=True) matricule_fiscal = fields.Char(required=True, index=True) business_type = fields.Selection( [("restaurant", "Restaurant"), ("fast_food", "Fast food / Sandwicherie"), ("cafe", "Café / Salon de thé"), ("bakery", "Pâtisserie / Glacier"), ("bar", "Bar / Buvette"), ("hotel", "Hôtel / Restaurant classé"), ("store", "Store / Retail"), ("other", "Other")], default="restaurant", required=True, help="Drives the POS preset provisioned for the tenant and the default " "A4 activity/VAT mapping.") pos_preset = fields.Char( compute="_compute_pos_preset", help="POS features provisioned for this business type.") rep_name = fields.Char(string="Legal representative") rep_cin = fields.Char(string="CIN") rep_phone = fields.Char(string="Mobile") rep_email = fields.Char(string="Email") admin_email = fields.Char( string="Tenant admin email", help="Account that receives the tenant-admin (accès contribuable) " "invitation once the database is provisioned.") db_name = fields.Char(string="Tenant database") state = fields.Selection( [("draft", "Draft"), ("provisioning", "Provisioning"), ("active", "Active"), ("suspended", "Suspended"), ("offboarded", "Offboarded")], default="draft", required=True, index=True) onboarded_date = fields.Date(readonly=True) suspended_date = fields.Date(readonly=True) notes = fields.Text() imdf_ids = fields.One2many("nacef.tenant.imdf", "tenant_id", string="IMDFs") agent_ids = fields.One2many("nacef.tenant.agent", "tenant_id", string="S-MDF Agents") imdf_count = fields.Integer(compute="_compute_counts") agent_count = fields.Integer(compute="_compute_counts") imdf_alert_count = fields.Integer( compute="_compute_health", help="IMDFs whose last reported S-MDF status is not SYNCHRONIZED.") cert_expiring_count = fields.Integer( compute="_compute_health", help="Certificates expiring within 30 days (E0606).") _sql_constraints = [ ("matricule_uniq", "unique(matricule_fiscal)", "A tenant with this matricule fiscal already exists."), ("db_uniq", "unique(db_name)", "This database name is already assigned to another tenant."), ] # POS features suggested per business type (drives provisioning). _POS_PRESETS = { "restaurant": "Tables & floors, kitchen printer, split bills, tips", "fast_food": "Fast counter, product screen, no tables", "cafe": "Counter + optional tables, quick products", "bakery": "Weighing/units, quick products, no tables", "bar": "Counter, quick products, alcohol VAT 19%", "hotel": "Rooms/tables, restaurant classé, alcohol VAT 19%", "store": "Barcode retail, no tables", "other": "Generic POS", } @api.depends("business_type") def _compute_pos_preset(self): for rec in self: rec.pos_preset = self._POS_PRESETS.get(rec.business_type, "") @api.depends("imdf_ids", "agent_ids") def _compute_counts(self): for rec in self: rec.imdf_count = len(rec.imdf_ids) rec.agent_count = len(rec.agent_ids) @api.depends("imdf_ids.last_status", "imdf_ids.cert_expiring_soon") def _compute_health(self): for rec in self: rec.imdf_alert_count = len(rec.imdf_ids.filtered( lambda i: i.last_status and i.last_status != "SYNCHRONIZED")) rec.cert_expiring_count = len( rec.imdf_ids.filtered("cert_expiring_soon")) # ------------------------------------------------------------------ # # lifecycle # ------------------------------------------------------------------ # def _make_db_name(self): """Build a hostname-safe database name that doubles as the tenant's subdomain label (letters/digits, starts with a letter, unique).""" self.ensure_one() base = re.sub(r"[^a-z0-9]", "", (self.name or "tenant").lower()) if not base or not base[0].isalpha(): base = "t" + base base = base[:40] or "tenant" candidate, i = base, 1 while (db_service.exp_db_exist(candidate) or self.search_count([("db_name", "=", candidate), ("id", "!=", self.id)])): i += 1 candidate = "%s%d" % (base, i) return candidate def _provision_database(self): """Assign the tenant database / subdomain name. The heavy DB creation happens in :meth:`_create_tenant_database`.""" for rec in self: if not rec.db_name: rec.db_name = rec._make_db_name() @staticmethod def _gen_password(): return "%s%02d-%s-%02d" % ( secrets.choice("ABCDEFGHJKLMNPQRSTUVWXYZ"), secrets.randbelow(100), secrets.choice(_PW_WORDS), secrets.randbelow(100)) def _admin_login(self): self.ensure_one() return (self.admin_email or self.rep_email or "admin@%s" % (self.db_name or "tenant")) def _tenant_login_url(self): """Subdomain URL: visiting it lands straight on this tenant's login (db-filter ^%d$ maps the host's first label to the database).""" self.ensure_one() domain = (self.env["ir.config_parameter"].sudo().get_param( "nacef.tenant_base_domain") or "localhost:8069").strip("/") # A real deployment terminates TLS at the reverse proxy and redirects # :80, so an http:// link would only bounce. Local development runs # Odoo directly on localhost with no certificate, so keep http there. scheme = "http" if domain.startswith("localhost") else "https" return "%s://%s.%s/" % (scheme, self.db_name, domain) def _create_tenant_database(self): """Really provision the tenant: clone the NACEF template DB into ``db_name`` and set the tenant-admin login + a temporary password. Returns the credentials dict the operator hands to the tenant.""" self.ensure_one() template = (self.env["ir.config_parameter"].sudo().get_param( "nacef.tenant_template") or "nacef_tpl") db_name = self.db_name # Use exp_db_exist (checks PostgreSQL directly); list_dbs honours the # instance dbfilter and would hide the template / other tenant DBs. if not db_service.exp_db_exist(template): raise UserError( "Base modèle « %s » introuvable. Créez-la d'abord " "(installez la stack NACEF), ou réglez le paramètre système " "nacef.tenant_template." % template) if db_service.exp_db_exist(db_name): raise UserError( "La base « %s » existe déjà pour un autre locataire." % db_name) try: db_service.exp_duplicate_database(template, db_name) except Exception as exc: # noqa: BLE001 _logger.exception("Tenant DB provisioning failed") raise UserError( "Échec de la création de la base « %s » : %s" % (db_name, exc) ) from exc login, password = self._admin_login(), self._gen_password() registry = odoo.modules.registry.Registry(db_name) with registry.cursor() as cr: env = api.Environment(cr, SUPERUSER_ID, {}) admin = env.ref("base.user_admin") admin.write({"login": login, "password": password, "email": self.admin_email or self.rep_email or False, "name": self.rep_name or admin.name}) admin.company_id.write({ "name": self.name, "nacef_matricule_fiscal": self.matricule_fiscal, }) cr.commit() return {"login": login, "password": password, "url": self._tenant_login_url()} def action_provision(self): creds = None for rec in self: if rec.state != "draft": raise UserError("Only a draft tenant can be provisioned.") rec.state = "provisioning" rec._provision_database() # Real DB creation is skipped inside the test runner and when the # caller opts out (context flag), so unit tests stay hermetic. testing = getattr(threading.current_thread(), "testing", False) if not testing and not self.env.context.get("nacef_skip_db_create"): creds = rec._create_tenant_database() rec.write({"state": "active", "onboarded_date": fields.Date.today()}) if creds and len(self) == 1: wizard = self.env["nacef.tenant.credentials"].create({ "tenant_id": self.id, "login_url": creds["url"], "admin_login": creds["login"], "temp_password": creds["password"], }) return { "type": "ir.actions.act_window", "name": "Accès locataire créé", "res_model": "nacef.tenant.credentials", "res_id": wizard.id, "view_mode": "form", "target": "new", } return True def action_suspend(self): for rec in self: if rec.state != "active": raise UserError("Only an active tenant can be suspended.") rec.write({"state": "suspended", "suspended_date": fields.Date.today()}) return True def action_reactivate(self): for rec in self: if rec.state != "suspended": raise UserError("Only a suspended tenant can be reactivated.") rec.state = "active" return True def action_offboard(self): for rec in self: if rec.state == "offboarded": raise UserError("This tenant is already offboarded.") # Real offboarding must export/archive the tenant DB first # (retention E1501); that is handled by the orchestrator. rec.state = "offboarded" return True