"""Accesul la date: componente, retete (produse finite) si pachete cantarite.""" from __future__ import annotations import math import re from datetime import datetime from typing import Any, Iterable, Optional from .db import get_db from .units import UNITS, convert, format_weight, normalize_unit MAX_RECIPE_COMPONENTS = 4 class ValidationError(ValueError): """Date introduse de utilizator invalide.""" def _now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") def _text(value: Any, limit: int) -> str: return str(value or "").strip()[:limit] def _number( value: Any, *, label: str, required: bool = False, minimum: float = 0.0, ) -> Optional[float]: if value in (None, ""): if required: raise ValidationError(f"{label} este obligatorie.") return None if isinstance(value, str): value = value.strip().replace(",", ".") try: number = float(value) except (TypeError, ValueError) as exc: raise ValidationError(f"{label} trebuie sa fie un numar.") from exc if not math.isfinite(number) or number < minimum: raise ValidationError(f"{label} trebuie sa fie un numar pozitiv.") return round(number, 3) def _flag(value: Any) -> int: return 1 if value in (True, 1, "1", "on", "true", "yes") else 0 def _optional_id(value: Any) -> Optional[int]: """Id numeric optional: '', None si 0 devin None; altfel int sau eroare.""" if value in (None, "", 0, "0"): return None if isinstance(value, bool): raise ValidationError("Identificator invalid.") try: number = int(value) except (TypeError, ValueError) as exc: raise ValidationError("Identificator invalid.") from exc return number if number > 0 else None # --------------------------------------------------------------------------- # Componente (ce se pune pe cantar) # --------------------------------------------------------------------------- def list_components(include_inactive: bool = False) -> list[dict[str, Any]]: sql = "SELECT * FROM components" if not include_inactive: sql += " WHERE active = 1" sql += " ORDER BY name COLLATE NOCASE" return [dict(r) for r in get_db().execute(sql)] def get_component(component_id: int) -> Optional[dict[str, Any]]: row = get_db().execute("SELECT * FROM components WHERE id = ?", (component_id,)).fetchone() return dict(row) if row else None def get_component_by_code(code: str) -> Optional[dict[str, Any]]: row = get_db().execute( "SELECT * FROM components WHERE code = ? COLLATE NOCASE", (_text(code, 40),) ).fetchone() return dict(row) if row else None def _clean_component(data: dict[str, Any], component_id: Optional[int] = None) -> dict[str, Any]: code = _text(data.get("code"), 40) name = _text(data.get("name"), 120) unit = normalize_unit(data.get("unit"), "g") if not code: raise ValidationError("Codul componentului este obligatoriu.") if not name: raise ValidationError("Denumirea componentului este obligatorie.") if unit not in UNITS: raise ValidationError(f"Unitate de masura necunoscuta: {unit}.") db = get_db() sql = "SELECT id FROM components WHERE code = ? COLLATE NOCASE" params: list[Any] = [code] if component_id is not None: sql += " AND id <> ?" params.append(component_id) if db.execute(sql, params).fetchone(): raise ValidationError(f"Exista deja un component cu codul {code!r}.") return { "code": code, "name": name, "unit": unit, "description": _text(data.get("description"), 500), "active": _flag(data.get("active", True)), } def create_component(data: dict[str, Any]) -> int: item = _clean_component(data) db = get_db() cur = db.execute( "INSERT INTO components(code, name, unit, description, active)" " VALUES (?, ?, ?, ?, ?)", (item["code"], item["name"], item["unit"], item["description"], item["active"]), ) db.commit() return int(cur.lastrowid) def update_component(component_id: int, data: dict[str, Any]) -> None: if get_component(component_id) is None: raise ValidationError("Componentul nu exista.") item = _clean_component(data, component_id) db = get_db() db.execute( "UPDATE components SET code = ?, name = ?, unit = ?, description = ?, active = ?," " updated_at = datetime('now') WHERE id = ?", (item["code"], item["name"], item["unit"], item["description"], item["active"], component_id), ) db.commit() def delete_component(component_id: int) -> None: """Sterge componentul daca nu este folosit; altfel il dezactiveaza.""" db = get_db() used = db.execute( "SELECT COUNT(*) FROM recipe_components WHERE component_id = ?", (component_id,) ).fetchone()[0] if used: db.execute( "UPDATE components SET active = 0, updated_at = datetime('now') WHERE id = ?", (component_id,), ) else: db.execute("DELETE FROM components WHERE id = ?", (component_id,)) db.commit() # --------------------------------------------------------------------------- # Retete (produsul finit) # --------------------------------------------------------------------------- def _recipe_components(recipe_id: int) -> list[dict[str, Any]]: rows = get_db().execute( "SELECT rc.*, c.code AS component_code, c.unit AS component_unit" " FROM recipe_components rc LEFT JOIN components c ON c.id = rc.component_id" " WHERE rc.recipe_id = ? ORDER BY rc.position", (recipe_id,), ) return [ { "position": row["position"], "component_id": row["component_id"], "code": row["component_code"] or "", "name": row["name"] or (row["component_code"] or ""), "target_weight": row["target_weight"], "unit": row["component_unit"] or "", } for row in rows ] def list_recipes(include_inactive: bool = False) -> list[dict[str, Any]]: sql = ( "SELECT r.*, t.name AS template_name," " (SELECT COUNT(*) FROM recipe_components rc WHERE rc.recipe_id = r.id) AS component_count" " FROM recipes r LEFT JOIN label_templates t ON t.id = r.template_id" ) if not include_inactive: sql += " WHERE r.active = 1" sql += " ORDER BY r.name COLLATE NOCASE" return [dict(r) for r in get_db().execute(sql)] def get_recipe(recipe_id: int) -> Optional[dict[str, Any]]: row = get_db().execute( "SELECT r.*, t.name AS template_name FROM recipes r" " LEFT JOIN label_templates t ON t.id = r.template_id WHERE r.id = ?", (recipe_id,), ).fetchone() if row is None: return None recipe = dict(row) recipe["components"] = _recipe_components(recipe_id) recipe["component_count"] = len(recipe["components"]) return recipe def _parse_recipe_components(data: dict[str, Any]) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] seen: set[int] = set() for slot in range(1, MAX_RECIPE_COMPONENTS + 1): raw_id = data.get(f"component_{slot}_id") if raw_id in (None, "", "0", 0): continue try: component_id = int(raw_id) except (TypeError, ValueError) as exc: raise ValidationError(f"Componentul {slot} nu este valid.") from exc component = get_component(component_id) if component is None or not component["active"]: raise ValidationError(f"Componentul {slot} nu exista sau este inactiv.") if component_id in seen: raise ValidationError(f"Componentul {component['name']!r} apare de doua ori.") seen.add(component_id) items.append({ "component_id": component_id, "code": component["code"], "name": component["name"], "target_weight": _number( data.get(f"component_{slot}_target"), label=f"Greutatea tinta pentru {component['name']}", ), }) if not items: raise ValidationError("O reteta are nevoie de cel putin un component.") for position, item in enumerate(items, start=1): item["position"] = position return items def _clean_recipe(data: dict[str, Any], recipe_id: Optional[int] = None) -> dict[str, Any]: code = _text(data.get("code"), 40) name = _text(data.get("name"), 120) unit = normalize_unit(data.get("unit"), "g") if not code: raise ValidationError("Codul produsului finit este obligatoriu.") if not name: raise ValidationError("Denumirea produsului finit este obligatorie.") if unit not in UNITS: raise ValidationError(f"Unitate de masura necunoscuta: {unit}.") db = get_db() sql = "SELECT id FROM recipes WHERE code = ? COLLATE NOCASE" params: list[Any] = [code] if recipe_id is not None: sql += " AND id <> ?" params.append(recipe_id) if db.execute(sql, params).fetchone(): raise ValidationError(f"Exista deja un produs finit cu codul {code!r}.") template_id = _optional_id(data.get("template_id")) if template_id is not None and get_template(template_id) is None: raise ValidationError("Sablonul de eticheta ales nu exista.") return { "code": code, "name": name, "unit": unit, "target_weight": _number(data.get("target_weight"), label="Greutatea totala tinta"), "note": _text(data.get("note"), 500), "active": _flag(data.get("active", True)), "template_id": template_id, "components": _parse_recipe_components(data), } def create_recipe(data: dict[str, Any]) -> int: recipe = _clean_recipe(data) db = get_db() cur = db.execute( "INSERT INTO recipes(code, name, unit, target_weight, note, active, template_id)" " VALUES (?, ?, ?, ?, ?, ?, ?)", (recipe["code"], recipe["name"], recipe["unit"], recipe["target_weight"], recipe["note"], recipe["active"], recipe["template_id"]), ) recipe_id = int(cur.lastrowid) _write_recipe_components(recipe_id, recipe["components"]) db.commit() return recipe_id def update_recipe(recipe_id: int, data: dict[str, Any]) -> None: if get_recipe(recipe_id) is None: raise ValidationError("Produsul finit nu exista.") recipe = _clean_recipe(data, recipe_id) db = get_db() db.execute( "UPDATE recipes SET code = ?, name = ?, unit = ?, target_weight = ?, note = ?," " active = ?, template_id = ?, updated_at = datetime('now') WHERE id = ?", (recipe["code"], recipe["name"], recipe["unit"], recipe["target_weight"], recipe["note"], recipe["active"], recipe["template_id"], recipe_id), ) db.execute("DELETE FROM recipe_components WHERE recipe_id = ?", (recipe_id,)) _write_recipe_components(recipe_id, recipe["components"]) db.commit() def _write_recipe_components(recipe_id: int, items: list[dict[str, Any]]) -> None: db = get_db() for item in items: db.execute( "INSERT INTO recipe_components(recipe_id, position, component_id, name, target_weight)" " VALUES (?, ?, ?, ?, ?)", (recipe_id, item["position"], item["component_id"], item["name"], item["target_weight"]), ) def delete_recipe(recipe_id: int) -> None: """Sterge reteta daca nu are pachete; altfel o dezactiveaza.""" db = get_db() used = db.execute( "SELECT COUNT(*) FROM packages WHERE recipe_id = ?", (recipe_id,) ).fetchone()[0] if used: db.execute("UPDATE recipes SET active = 0, updated_at = datetime('now') WHERE id = ?", (recipe_id,)) else: db.execute("DELETE FROM recipes WHERE id = ?", (recipe_id,)) db.commit() # --------------------------------------------------------------------------- # Pachete (o eticheta de produs finit) # --------------------------------------------------------------------------- def _package_row(package_id: int) -> Optional[dict[str, Any]]: row = get_db().execute("SELECT * FROM packages WHERE id = ?", (package_id,)).fetchone() return dict(row) if row else None def get_package(package_id: int) -> Optional[dict[str, Any]]: package = _package_row(package_id) if package is None: return None package["items"] = [ dict(row) for row in get_db().execute( "SELECT * FROM package_items WHERE package_id = ? ORDER BY position", (package_id,), ) ] return package def get_open_package() -> Optional[dict[str, Any]]: row = get_db().execute( "SELECT id FROM packages WHERE status = 'open' ORDER BY id DESC LIMIT 1" ).fetchone() return get_package(row["id"]) if row else None def start_package( recipe_id: int, *, operator: str = "", lot: str = "", notes: str = "", template_id: Optional[int] = None, ) -> int: recipe = get_recipe(recipe_id) if recipe is None or not recipe["active"]: raise ValidationError("Produsul finit nu exista sau este inactiv.") if not recipe["components"]: raise ValidationError("Produsul finit nu are componente configurate.") if get_open_package() is not None: raise ValidationError("Exista deja un pachet inceput. Finalizeaza-l sau renunta la el.") db = get_db() cur = db.execute( "INSERT INTO packages(created_at, status, recipe_id, recipe_code, recipe_name, unit," " target_weight, operator, lot, notes, template_id)" " VALUES (?, 'open', ?, ?, ?, ?, ?, ?, ?, ?, ?)", (_now_iso(), recipe["id"], recipe["code"], recipe["name"], recipe["unit"], recipe["target_weight"], _text(operator, 64), _text(lot, 64), _text(notes, 500), template_id), ) db.commit() return int(cur.lastrowid) def add_package_item( package_id: int, *, weight: float, unit: str, source: str = "manual", stable: bool = True, scale_profile: str = "", raw_hex: str = "", ) -> dict[str, Any]: """Adauga urmatoarea cantarire din reteta si intoarce pachetul actualizat.""" package = get_package(package_id) if package is None: raise ValidationError("Pachetul nu exista.") if package["status"] != "open": raise ValidationError("Pachetul este deja finalizat.") recipe = get_recipe(package["recipe_id"]) if package["recipe_id"] else None plan = recipe["components"] if recipe else [] position = len(package["items"]) + 1 if position > len(plan): raise ValidationError("Toate componentele au fost cantarite.") planned = plan[position - 1] normalized = normalize_unit(unit, package["unit"]) converted = convert(float(weight), normalized, package["unit"]) if converted is None: raise ValidationError(f"Unitatea {normalized!r} nu poate fi convertita in {package['unit']!r}.") value = round(converted, 3) if value <= 0: raise ValidationError("Greutatea trebuie sa fie mai mare decat zero.") db = get_db() db.execute( "INSERT INTO package_items(package_id, position, component_id, component_code," " component_name, weight, unit, target_weight, stable, source, scale_profile, raw_hex," " created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (package_id, position, planned.get("component_id"), planned.get("code") or "", planned.get("name") or "", value, package["unit"], planned.get("target_weight"), 1 if stable else 0, "scale" if source == "scale" else "manual", _text(scale_profile, 64), _text(raw_hex, 500), _now_iso()), ) db.commit() return get_package(package_id) or package def undo_package_item(package_id: int) -> dict[str, Any]: package = get_package(package_id) if package is None: raise ValidationError("Pachetul nu exista.") if package["status"] != "open": raise ValidationError("Pachetul este finalizat; nu se mai poate reveni.") if not package["items"]: raise ValidationError("Nu exista nicio cantarire de anulat.") db = get_db() db.execute( "DELETE FROM package_items WHERE package_id = ? AND position = ?", (package_id, max(item["position"] for item in package["items"])), ) db.commit() return get_package(package_id) or package def complete_package(package_id: int) -> dict[str, Any]: package = get_package(package_id) if package is None: raise ValidationError("Pachetul nu exista.") if package["status"] == "done": return package if package["status"] != "open": raise ValidationError("Pachetul nu mai poate fi finalizat.") recipe = get_recipe(package["recipe_id"]) if package["recipe_id"] else None planned = len(recipe["components"]) if recipe else 0 if len(package["items"]) != planned or planned == 0: raise ValidationError("Nu toate componentele au fost cantarite.") total = round(sum(float(item["weight"]) for item in package["items"]), 3) db = get_db() db.execute( "UPDATE packages SET status = 'done', completed_at = ?, total_weight = ? WHERE id = ?", (_now_iso(), total, package_id), ) db.commit() return get_package(package_id) or package def void_package(package_id: int) -> None: db = get_db() db.execute( "UPDATE packages SET status = 'void', completed_at = COALESCE(completed_at, ?)" " WHERE id = ? AND status = 'open'", (_now_iso(), package_id), ) db.commit() def mark_printed(package_id: int) -> Optional[dict[str, Any]]: db = get_db() db.execute( "UPDATE packages SET printed_at = ?, print_count = print_count + 1 WHERE id = ?", (_now_iso(), package_id), ) db.commit() return get_package(package_id) def package_state(package_id: int) -> Optional[dict[str, Any]]: """Starea fluxului: pachet, reteta, pasul curent si totalul.""" package = get_package(package_id) if package is None: return None recipe = get_recipe(package["recipe_id"]) if package["recipe_id"] else None plan = recipe["components"] if recipe else [] done = len(package["items"]) step = None if package["status"] == "open" and done < len(plan): planned = plan[done] step = { "position": done + 1, "total": len(plan), "component_id": planned.get("component_id"), "component_code": planned.get("code") or "", "component_name": planned.get("name") or "", "target_weight": planned.get("target_weight"), "unit": package["unit"], } return { "package": package, "recipe": recipe, "items": package["items"], "step": step, "planned_total": len(plan), "total_weight": round(sum(float(item["weight"]) for item in package["items"]), 3), "complete": package["status"] == "done", "open": package["status"] == "open", } def list_packages( *, date_from: str = "", date_to: str = "", recipe_id: Optional[int] = None, status: str = "", query: str = "", limit: int = 500, ) -> list[dict[str, Any]]: sql = ( "SELECT p.*," " (SELECT COUNT(*) FROM package_items i WHERE i.package_id = p.id) AS item_count," " (SELECT GROUP_CONCAT(i.component_name || ' ' || printf('%g', i.weight) || ' ' || i.unit, '; ')" " FROM package_items i WHERE i.package_id = p.id) AS item_summary" " FROM packages p WHERE 1 = 1" ) params: list[Any] = [] if date_from: sql += " AND COALESCE(p.completed_at, p.created_at) >= ?" params.append(date_from) if date_to: sql += " AND COALESCE(p.completed_at, p.created_at) <= ?" params.append(date_to + "T23:59:59.999999" if len(date_to) == 10 else date_to) if recipe_id: sql += " AND p.recipe_id = ?" params.append(recipe_id) if status: sql += " AND p.status = ?" params.append(status) if query: sql += (" AND (p.recipe_name LIKE ? OR p.recipe_code LIKE ? OR p.lot LIKE ?" " OR p.operator LIKE ?)") like = f"%{query}%" params.extend([like, like, like, like]) sql += " ORDER BY COALESCE(p.completed_at, p.created_at) DESC, p.id DESC LIMIT ?" params.append(max(1, min(int(limit), 5000))) return [dict(row) for row in get_db().execute(sql, params)] def recent_packages(limit: int = 6) -> list[dict[str, Any]]: return list_packages(limit=limit) def package_stats() -> dict[str, Any]: db = get_db() today = db.execute( "SELECT COUNT(*) FROM packages WHERE status = 'done'" " AND date(COALESCE(completed_at, created_at)) = date('now', 'localtime')" ).fetchone()[0] units = db.execute( "SELECT unit, SUM(total_weight) AS total FROM packages WHERE status = 'done'" " AND date(COALESCE(completed_at, created_at)) = date('now', 'localtime')" " GROUP BY unit ORDER BY unit" ).fetchall() recipes = db.execute("SELECT COUNT(*) FROM recipes WHERE active = 1").fetchone()[0] return { "today_count": today, "today_total": ", ".join( f"{format_weight(row['total'] or 0)} {row['unit']}" for row in units ) or "0", "recipes": recipes, } def to_csv_rows(packages: Iterable[dict[str, Any]]) -> list[list[str]]: rows = [[ "ID", "Data", "Cod produs finit", "Produs finit", "Lot", "Operator", "Greutate totala", "UM", "Componente", "Status", "Tiparit la", "Numar tipariri", ]] for package in packages: rows.append([ str(package["id"]), package.get("completed_at") or package.get("created_at") or "", package.get("recipe_code") or "", package.get("recipe_name") or "", package.get("lot") or "", package.get("operator") or "", f"{float(package.get('total_weight') or 0):.3f}", package.get("unit") or "", package.get("item_summary") or "", package.get("status") or "", package.get("printed_at") or "", str(package.get("print_count") or 0), ]) return rows def slugify(text: str) -> str: text = re.sub(r"[^A-Za-z0-9_-]+", "-", text.strip().lower()) return text.strip("-_") or "eticheta" # --------------------------------------------------------------------------- # Sabloane de eticheta # --------------------------------------------------------------------------- def list_templates() -> list[dict[str, Any]]: return [dict(row) for row in get_db().execute( "SELECT * FROM label_templates ORDER BY is_default DESC, name COLLATE NOCASE" )] def get_template(template_id: int) -> Optional[dict[str, Any]]: row = get_db().execute( "SELECT * FROM label_templates WHERE id = ?", (template_id,) ).fetchone() return dict(row) if row else None def get_default_template() -> Optional[dict[str, Any]]: row = get_db().execute( "SELECT * FROM label_templates ORDER BY is_default DESC, id LIMIT 1" ).fetchone() return dict(row) if row else None def _clean_template(data: dict[str, Any], template_id: Optional[int] = None) -> dict[str, Any]: name = _text(data.get("name"), 80) # `body_html`/`css` sunt coloane vechi (modul HTML/CSS scos); le pastram goale # pentru compatibilitatea schemei, fara sa cerem continut de la utilizator. body = str(data.get("body_html") or "").strip() css = str(data.get("css") or "").strip() design_json = data.get("design_json") design_json = str(design_json) if design_json is not None else None if not name: raise ValidationError("Numele sablonului este obligatoriu.") if not body and not design_json: raise ValidationError("Continutul etichetei este obligatoriu.") width = _number(data.get("width_mm"), label="Latimea etichetei") or 100.0 height = _number(data.get("height_mm"), label="Inaltimea etichetei") or 60.0 if not (10 <= width <= 500 and 10 <= height <= 500): raise ValidationError("Dimensiunile etichetei trebuie sa fie intre 10 si 500 mm.") db = get_db() sql = "SELECT id FROM label_templates WHERE name = ? COLLATE NOCASE" params: list[Any] = [name] if template_id is not None: sql += " AND id <> ?" params.append(template_id) if db.execute(sql, params).fetchone(): raise ValidationError(f"Exista deja un sablon numit {name!r}.") return { "name": name, "width_mm": width, "height_mm": height, "body_html": body, "css": css, "design_json": design_json, "is_default": _flag(data.get("is_default")), } def create_template(data: dict[str, Any]) -> int: template = _clean_template(data) db = get_db() if template["is_default"]: db.execute("UPDATE label_templates SET is_default = 0") cur = db.execute( "INSERT INTO label_templates(name, width_mm, height_mm, body_html, css, design_json," " is_default) VALUES (?, ?, ?, ?, ?, ?, ?)", (template["name"], template["width_mm"], template["height_mm"], template["body_html"], template["css"], template["design_json"] or "", template["is_default"]), ) db.commit() return int(cur.lastrowid) def update_template(template_id: int, data: dict[str, Any]) -> None: existing = get_template(template_id) if existing is None: raise ValidationError("Sabloul nu exista.") payload = dict(data) # Formularul HTML nu trimite design; pastram designul existent. if "design_json" not in payload: payload["design_json"] = existing.get("design_json") or "" template = _clean_template(payload, template_id) db = get_db() if template["is_default"]: db.execute("UPDATE label_templates SET is_default = 0") db.execute( "UPDATE label_templates SET name = ?, width_mm = ?, height_mm = ?, body_html = ?," " css = ?, design_json = ?, is_default = ?, updated_at = datetime('now') WHERE id = ?", (template["name"], template["width_mm"], template["height_mm"], template["body_html"], template["css"], template["design_json"] or "", template["is_default"], template_id), ) db.commit() def save_template_design(template_id: int, design_json: str) -> None: """Salveaza designul vizual (JSON) pentru un sablon.""" if get_template(template_id) is None: raise ValidationError("Sabloul nu exista.") db = get_db() db.execute( "UPDATE label_templates SET design_json = ?, updated_at = datetime('now') WHERE id = ?", (design_json, template_id), ) db.commit() def delete_template(template_id: int) -> None: db = get_db() template = get_template(template_id) if template is None: return used = db.execute( "SELECT COUNT(*) FROM packages WHERE template_id = ?", (template_id,) ).fetchone()[0] if used: raise ValidationError("Sabloul este folosit de pachete existente si nu poate fi sters.") bound = db.execute( "SELECT COUNT(*) FROM recipes WHERE template_id = ? AND active = 1", (template_id,) ).fetchone()[0] if bound: raise ValidationError( "Sabloul este atribuit unor produse finite active. Schimba sablonul acestora" " inainte de stergere." ) db.execute("DELETE FROM label_templates WHERE id = ?", (template_id,)) if template["is_default"]: row = db.execute("SELECT id FROM label_templates ORDER BY id LIMIT 1").fetchone() if row: db.execute("UPDATE label_templates SET is_default = 1 WHERE id = ?", (row["id"],)) db.commit() # --------------------------------------------------------------------------- # Campuri variabile proprii (custom.*) # --------------------------------------------------------------------------- CUSTOM_KEY_PREFIX = "custom." # Diacriticele romanesti nu se descompun in NFKD (s/ț au virgula dedesubt), # deci le transliteram explicit inainte de a construi cheia. _KEY_TRANSLIT = str.maketrans({ "ă": "a", "â": "a", "á": "a", "à": "a", "ä": "a", "å": "a", "î": "i", "í": "i", "ì": "i", "ï": "i", "ș": "s", "ş": "s", "ś": "s", "ț": "t", "ţ": "t", "é": "e", "è": "e", "ë": "e", "ó": "o", "ò": "o", "ö": "o", "ú": "u", "ù": "u", "ü": "u", "ç": "c", "ñ": "n", }) def _key_slug(text: str) -> str: lowered = str(text or "").strip().lower().translate(_KEY_TRANSLIT) return re.sub(r"[^a-z0-9]+", "_", lowered).strip("_")[:40] def list_label_fields(include_inactive: bool = False) -> list[dict[str, Any]]: sql = "SELECT * FROM label_fields" if not include_inactive: sql += " WHERE active = 1" sql += " ORDER BY label COLLATE NOCASE" return [dict(row) for row in get_db().execute(sql)] def get_label_field(field_id: int) -> Optional[dict[str, Any]]: row = get_db().execute( "SELECT * FROM label_fields WHERE id = ?", (field_id,) ).fetchone() return dict(row) if row else None def get_label_field_by_key(key: str) -> Optional[dict[str, Any]]: row = get_db().execute( "SELECT * FROM label_fields WHERE key = ?", (key,) ).fetchone() return dict(row) if row else None def _new_field_key(label: str) -> str: """`Contact firmă` -> `custom.contact_firma`, cu sufix numeric daca exista deja.""" base = _key_slug(label) or "camp" key = f"{CUSTOM_KEY_PREFIX}{base}" db = get_db() counter = 2 while db.execute("SELECT id FROM label_fields WHERE key = ?", (key,)).fetchone(): key = f"{CUSTOM_KEY_PREFIX}{base}_{counter}" counter += 1 return key def _clean_label_field(data: dict[str, Any]) -> dict[str, Any]: label = _text(data.get("label"), 60) template = str(data.get("template") or "").strip() if not label: raise ValidationError("Denumirea campului este obligatorie.") if not template: raise ValidationError("Textul campului este obligatoriu.") if len(template) > 300: raise ValidationError("Textul campului poate avea maxim 300 de caractere.") return {"label": label, "template": template, "active": _flag(data.get("active"))} def create_label_field(data: dict[str, Any]) -> int: field = _clean_label_field(data) key = _new_field_key(field["label"]) db = get_db() cur = db.execute( "INSERT INTO label_fields(key, label, template, active) VALUES (?, ?, ?, ?)", (key, field["label"], field["template"], field["active"]), ) db.commit() return int(cur.lastrowid) def update_label_field(field_id: int, data: dict[str, Any]) -> None: if get_label_field(field_id) is None: raise ValidationError("Campul nu exista.") field = _clean_label_field(data) db = get_db() # Cheia ramane neschimbata: designurile o refera deja. db.execute( "UPDATE label_fields SET label = ?, template = ?, active = ?," " updated_at = datetime('now') WHERE id = ?", (field["label"], field["template"], field["active"], field_id), ) db.commit() def count_label_field_usage(key: str) -> int: """In cate sabloane (design) apare campul dat.""" needle = f'"{key}"' row = get_db().execute( "SELECT COUNT(*) FROM label_templates WHERE design_json LIKE ?", (f"%{needle}%",) ).fetchone() return int(row[0]) if row else 0 def delete_label_field(field_id: int) -> None: db = get_db() db.execute("DELETE FROM label_fields WHERE id = ?", (field_id,)) db.commit()