#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Générateur déterministe des modèles physiques HFSQL Omika. Lit docs/schema-compatibilite-batik-omika.json et génère les classes C# des modèles physiques Batik (Isis_BO) et Omika dans src/Omika.Data. Ce script est un outil interne : - la génération est déterministe (même entrée => même sortie) ; - il n'est pas nécessaire au fonctionnement de l'application ; - il n'invente aucune donnée : chaque propriété provient du JSON ; - la compilation ne dépend pas de ce script ni d'un chemin absolu. Fichiers générés : src/Omika.Data/HfSql/Models/Batik/Row.cs src/Omika.Data/HfSql/Models/Omika/
Row.cs """ from __future__ import annotations import json import re import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent JSON_PATH = ROOT / "docs" / "schema-compatibilite-batik-omika.json" OUT_BATIK = ROOT / "src" / "Omika.Data" / "HfSql" / "Models" / "Batik" OUT_OMIKA = ROOT / "src" / "Omika.Data" / "HfSql" / "Models" / "Omika" HEADER = ( "// \n" "// Généré à partir de docs/schema-compatibilite-batik-omika.json.\n" "// Ne pas modifier à la main.\n" "#nullable enable\n" ) # Types HFSQL -> type C# par défaut (quand le JSON ne recommande rien d'autre). HF_TYPES = { "Identifiant automatique (8 octets)": "long", "Entier sur 8 octets": "long", "Entier non signé sur 8 octets": "ulong", "Entier sur 4 octets": "int", "Entier non signé sur 4 octets": "uint", "Entier sur 2 octets": "short", "Entier non signé sur 2 octets": "ushort", "Entier sur 1 octet": "sbyte", "Entier non signé sur 1 octet": "byte", "Booléen": "bool", "Chaîne": "string", "Chaîne ANSI": "string", "Chaîne Unicode": "string", "Mémo texte": "string", "Mémo binaire": "byte[]", "Image (mémo binaire)": "byte[]", "Date (aaaammjj)": "DateOnly", "Heure (hhmm)": "TimeOnly", "Date et Heure (aaaammjjhhmmssccc)": "DateTime", "Réel sur 4 octets": "float", "Réel sur 8 octets": "double", "Numérique": "double", "Monétaire": "decimal", "Clé composée": "string", "Durée": "TimeSpan", } def to_pascal(name: str) -> str: """Convertit un nom de rubrique HFSQL en nom C# PascalCase.""" parts = re.split(r"_+", name) return "".join(p[:1].upper() + p[1:] for p in parts if p) def to_identifier(name: str) -> str: """Transforme un nom de table HFSQL en identifiant C#.""" ident = to_pascal(name) if ident and ident[0].isdigit(): ident = "T" + ident # Évite les mots-clés C# keywords = {"class", "namespace", "string", "object", "bool", "int", "long"} if ident.lower() in keywords: ident += "_" return ident def net_type(field: dict, hfsql_type: str | None) -> str: """Détermine le type .NET pour une rubrique.""" rec = field.get("dotnet_type_recommendation") if rec: if rec == "string (32) en persistance; Guid? après validation": return "string" if rec == "string en lecture seule / index technique": return "string" return rec if hfsql_type and hfsql_type in HF_TYPES: return HF_TYPES[hfsql_type] return "string" # Construit les entités XML via chr() pour éviter toute interprétation # des séquences d'échappement dans ce générateur. _XML_AMP = chr(38) # & _XML_LT = chr(60) # < _XML_GT = chr(62) # > _XML_QUOT = chr(34) # " def escape_cs(s: str) -> str: """Échappe une chaîne pour l'écrire dans une doc XML C#.""" return ( s.replace(_XML_AMP, _XML_AMP + "amp;") .replace(_XML_LT, _XML_LT + "lt;") .replace(_XML_GT, _XML_GT + "gt;") .replace(_XML_QUOT, _XML_QUOT + "quot;") ) def _find_zombie_replacement(physical_name: str, fields: list) -> str | None: """Trouve une rubrique de remplacement documentée pour une rubrique zombie. La règle repose sur les conventions du schéma Omika : une rubrique numérique historique `XYZ_ID...` est remplacée par une clé textuelle `XYZ_Cle...` partageant le même noyau. Ex. `DC_Categorie` -> `DC_CleCategorie`, `DC_IDEtatDocument` -> `DC_CleEtatDoc`. """ core = re.sub(r"^(.*?)_(ID[A-Za-z0-9_]*|[A-Za-z0-9]*ID)", r"\1", physical_name) for other in fields: on = other.get("canonical_name") or "" if on and on != physical_name and ("Cle" in on or "GUID" in on): other_core = re.sub(r"^(.*?)_(Cle[A-Za-z0-9_]*|GUID)", r"\1", on) if core and other_core and (core in other_core or other_core in core): return on return None def build_class(table: dict, version: str) -> str: """Construit le code C# d'un modèle physique pour une version donnée.""" name = table["name"] ns = f"Omika.Data.HfSql.Models.{'Batik' if version == 'batik' else 'Omika'}" cls = to_identifier(name) + "Row" meta = table.get("old_metadata" if version == "batik" else "new_metadata") disk = meta.get("disk_name") if meta else None groups = (meta.get("groups") or []) if meta else [] schema_versions = ["Batik"] if version == "batik" else ["Omika"] lines = [HEADER, "", f"namespace {ns};", ""] # Pas de « using » dans les fichiers générés : les attributs de mapping sont # exposés via le « global using » du projet Omika.Data (GlobalUsings.cs), # ce qui évite la résolution relative délicate des usings dans les namespaces # « Omika.Data.HfSql.Models.* ». lines.append("/// ") lines.append(f"/// Modèle physique de la table {name} dans le schéma {schema_versions[0]}.") if table.get("purpose_inference"): lines.append(f"/// {escape_cs(table['purpose_inference'])}") lines.append("/// ") attr = f'[HfSqlTable("{name}"' if disk: attr += f', DiskName = "{disk}"' if groups: attr += f', Groups = [{", ".join(f'"{g}"' for g in groups)}]' attr += f', SchemaVersions = [{", ".join(f'"{v}"' for v in schema_versions)}])]' lines.append(attr) lines.append(f"public class {cls}") lines.append("{") lines.append(f" /// Constructeur sans paramètre (lecture HFSQL).") lines.append(f" public {cls}()") lines.append(f" {{") lines.append(f" }}") lines.append("") # Clé JSON du côté : « old » pour l'ancienne base (Isis_BO), « new » pour Omika. json_side = "old" if version == "batik" else "new" for fld in table["fields"]: side = fld.get(json_side) if side is None: # Rubrique absente de cette version continue physical_name = side.get("name") or fld.get("canonical_name") canonical = fld.get("canonical_name") label = side.get("label") hfsql_type = side.get("hfsql_type") size = side.get("size") default = side.get("default") zombie = side.get("zombie") is True notes = side.get("notes") or [] is_pk = "PK automatique" in (fld.get("compatibility_notes") or "") or ( "PK automatique" in (notes or []) ) is_composite = "Clé composée" in (hfsql_type or "") compat = fld.get("compatibility_notes") or "" prop_name = to_identifier(canonical) if not prop_name: prop_name = to_identifier(physical_name) prop_name = to_identifier(prop_name) # Résout un éventuel conflit de nom avec la classe if prop_name == cls: prop_name += "_" t = net_type(fld, hfsql_type) t = t.strip() lines.append(" /// ") if label: lines.append(f" /// {escape_cs(label)}") else: lines.append(f" /// Rubrique {escape_cs(physical_name)}.") if zombie: lines.append(" /// Rubrique historique (zombie) : conservée pour la lecture d'une base existante,") lines.append(" /// ne doit pas être utilisée comme référence principale dans le nouveau domaine.") if size: lines.append(f" /// Taille maximale : {escape_cs(size)}.") if default: lines.append(f" /// Valeur par défaut documentée : {escape_cs(default)}.") if notes: for n in notes: lines.append(f" /// Note : {escape_cs(n)}.") lines.append(" /// ") # Attributs col_attr = f' [HfSqlColumn("{physical_name}"' if hfsql_type: col_attr += f', HfSqlType = "{escape_cs(hfsql_type)}"' if size and size.isdigit(): col_attr += f", MaxLength = {int(size)}" if default: col_attr += f', DefaultValue = "{escape_cs(default)}"' if zombie: col_attr += ", IsZombie = true" if version == "omika" and side.get("name") is not None and fld.get("old") is None: col_attr += ", IsOmikaOnly = true" col_attr += ")]" lines.append(col_attr) if is_pk: lines.append(f' [HfSqlPrimaryKey("{physical_name}")]') if is_composite: # clé composée : on note les composantes si elles sont dans les notes comps = re.findall(r"([A-Za-z0-9_]+)\+([A-Za-z0-9_]+)", compat + " " + " ".join(notes)) if comps: cols = [c for pair in comps for c in pair] quoted = ", ".join('"' + c + '"' for c in cols) lines.append(" [HfSqlCompositeKey(" + quoted + ")]") if zombie: replacement = _find_zombie_replacement(physical_name, table["fields"]) if replacement: lines.append(f' [HfSqlLegacyField("{physical_name}", ReplacementColumn = "{replacement}")]') else: lines.append(f' [HfSqlLegacyField("{physical_name}")]') # Les types référence sont déclarés nullables : la nullabilité réelle n'est # pas démontrée par l'export, et on ne suppose pas qu'une chaîne vide est # une valeur valide. Les types valeur conservent le type recommandé par le JSON. if t in ("string", "byte[]"): t += "?" lines.append(f" public {t} {prop_name} {{ get; set; }}") lines.append("") lines.append("}") return "\n".join(lines) + "\n" def main() -> int: with open(JSON_PATH, encoding="utf-8") as f: data = json.load(f) tables = data["tables"] generated = [] for t in tables: for version, out_dir in (("batik", OUT_BATIK), ("omika", OUT_OMIKA)): meta = t.get("old_metadata" if version == "batik" else "new_metadata") if meta is None: continue out_dir.mkdir(parents=True, exist_ok=True) code = build_class(t, version) cls = to_identifier(t["name"]) + "Row" out = out_dir / f"{cls}.cs" out.write_text(code, encoding="utf-8") generated.append(out.relative_to(ROOT)) print(f"Généré {len(generated)} fichiers.") for g in sorted(str(x) for x in generated): print(" -", g) return 0 if __name__ == "__main__": sys.exit(main())