412 lines
12 KiB
Python
412 lines
12 KiB
Python
import marimo
|
||
|
||
__generated_with = "0.16.5"
|
||
app = marimo.App(width="medium")
|
||
|
||
|
||
@app.cell
|
||
def _():
|
||
import marimo as mo
|
||
return (mo,)
|
||
|
||
|
||
@app.cell
|
||
def _():
|
||
from pif_compiler.functions.db_utils import db_connect
|
||
return (db_connect,)
|
||
|
||
|
||
@app.cell
|
||
def _(db_connect):
|
||
col = db_connect(collection_name="orders")
|
||
return (col,)
|
||
|
||
|
||
@app.cell
|
||
def _(col):
|
||
input = col.find_one({"client_name": "CSM Srl"})
|
||
return (input,)
|
||
|
||
|
||
@app.cell
|
||
def _(input):
|
||
input
|
||
return
|
||
|
||
|
||
@app.cell
|
||
def _():
|
||
import json
|
||
from pydantic import BaseModel, Field, field_validator, ConfigDict, model_validator
|
||
from pymongo import MongoClient
|
||
import re
|
||
from typing import List, Optional
|
||
return BaseModel, ConfigDict, Field, List, Optional, model_validator, re
|
||
|
||
|
||
app._unparsable_cell(
|
||
r"""
|
||
|
||
|
||
class CosmeticIngredient(BaseModel):
|
||
inci_name: str
|
||
cas: str = Field(..., pattern=r'^\d{2,7}-\d{2}-\d$')
|
||
colorant = bool = Field(default=False)
|
||
organic = bool = Field(default=False)
|
||
|
||
dap = dict | None = Field(default=None)
|
||
cosing = dict | None = Field(default=None)
|
||
tox_levels = dict | None = Field(default=None)
|
||
|
||
@field_validator('inci_name')
|
||
@classmethod
|
||
def make_uppercase(cls, v: str) -> str:
|
||
return v.upper()
|
||
""",
|
||
name="_"
|
||
)
|
||
|
||
|
||
@app.cell
|
||
def _(CosmeticIngredient, collection, mo):
|
||
mo.stop(True)
|
||
try:
|
||
ingredient = CosmeticIngredient(
|
||
inci_name="Glycerin",
|
||
cas="56-81-5",
|
||
percentage=5.5
|
||
)
|
||
print(f"✅ Object Created: {ingredient}")
|
||
except ValueError as e:
|
||
print(f"❌ Validation Error: {e}")
|
||
|
||
document_to_insert = ingredient.model_dump()
|
||
|
||
result = collection.insert_one(document_to_insert)
|
||
return
|
||
|
||
|
||
@app.cell
|
||
def _(
|
||
BaseModel,
|
||
ConfigDict,
|
||
CosingInfo,
|
||
Field,
|
||
List,
|
||
Optional,
|
||
mo,
|
||
model_validator,
|
||
re,
|
||
):
|
||
mo.stop(True)
|
||
|
||
|
||
class DapInfo(BaseModel):
|
||
"""Informazioni dal Dossier (es. origine, purezza)"""
|
||
origin: Optional[str] = None # es. "Synthetic", "Vegetable"
|
||
purity_percentage: Optional[float] = None
|
||
supplier_code: Optional[str] = None
|
||
|
||
class ToxInfo(BaseModel):
|
||
"""Dati Tossicologici"""
|
||
noael: Optional[float] = None # No Observed Adverse Effect Level
|
||
ld50: Optional[float] = None # Lethal Dose 50
|
||
sed: Optional[float] = None # Systemic Exposure Dosage
|
||
mos: Optional[float] = None # Margin of Safety
|
||
|
||
# --- 2. Modello Principale ---
|
||
|
||
class CosmeticIngredient(BaseModel):
|
||
model_config = ConfigDict(validate_assignment=True) # Valida anche se modifichi i campi dopo
|
||
|
||
# Gestione INCI multipli per lo stesso CAS
|
||
inci_names: List[str] = Field(default_factory=list)
|
||
|
||
# Il CAS è una stringa obbligatoria, ma il regex dipende dal contesto
|
||
cas: str
|
||
|
||
colorant: bool = Field(default=False)
|
||
organic: bool = Field(default=False)
|
||
|
||
# Sotto-oggetti opzionali
|
||
dap: Optional[DapInfo] = None
|
||
cosing: Optional[CosingInfo] = None
|
||
tox_levels: Optional[ToxInfo] = None
|
||
|
||
# --- VALIDAZIONE CONDIZIONALE CAS ---
|
||
@model_validator(mode='after')
|
||
def validate_cas_logic(self):
|
||
cas_value = self.cas
|
||
is_exempt = self.colorant or self.organic
|
||
|
||
if not cas_value or not cas_value.strip():
|
||
raise ValueError("Il campo CAS non può essere vuoto.")
|
||
|
||
if not is_exempt:
|
||
cas_regex = r'^\d{2,7}-\d{2}-\d$'
|
||
if not re.match(cas_regex, cas_value):
|
||
raise ValueError(f"Formato CAS non valido ('{cas_value}') per ingrediente standard.")
|
||
|
||
# Se è colorante/organico, accettiamo qualsiasi stringa (es. 'CI 77891' o codici interni)
|
||
return self
|
||
|
||
# --- METODO HELPER PER AGGIUNGERE INCI ---
|
||
def add_inci(self, new_inci: str):
|
||
"""Aggiunge un INCI alla lista solo se non è già presente (case insensitive)."""
|
||
new_inci_upper = new_inci.upper()
|
||
# Controlliamo se esiste già (normalizzando a maiuscolo per sicurezza)
|
||
if not any(existing.upper() == new_inci_upper for existing in self.inci_names):
|
||
self.inci_names.append(new_inci_upper)
|
||
print(f"✅ INCI '{new_inci_upper}' aggiunto.")
|
||
else:
|
||
print(f"ℹ️ INCI '{new_inci_upper}' già presente.")
|
||
return CosmeticIngredient, DapInfo
|
||
|
||
|
||
@app.cell
|
||
def _():
|
||
from pif_compiler.services.srv_pubchem import pubchem_dap
|
||
|
||
dato = pubchem_dap("56-81-5")
|
||
dato
|
||
return (dato,)
|
||
|
||
|
||
app._unparsable_cell(
|
||
r"""
|
||
molecular_weight = dato.get(\"MolecularWeight\")
|
||
log pow = dato.get(\"XLogP\")
|
||
topological_polar_surface_area = dato.get(\"TPSA\")
|
||
melting_point = dato.get(\"MeltingPoint\")
|
||
ionization = dato.get(\"Dissociation Constants\")
|
||
""",
|
||
name="_"
|
||
)
|
||
|
||
|
||
@app.cell(hide_code=True)
|
||
def _(mo):
|
||
mo.md(
|
||
r"""
|
||
Molecolar Weight >500 Da
|
||
High degree of ionisation
|
||
Log Pow ≤-1 or ≥ 4
|
||
Topological polar surface area >120 Å2
|
||
Melting point > 200°C
|
||
"""
|
||
)
|
||
return
|
||
|
||
|
||
@app.cell
|
||
def _(BaseModel, Field, Optional, model_validator):
|
||
class DapInfo(BaseModel):
|
||
cas: str
|
||
|
||
molecular_weight: Optional[float] = Field(default=None, description="In Daltons (Da)")
|
||
high_ionization: Optional[float] = Field(default=None, description="High degree of ionization")
|
||
log_pow: Optional[float] = Field(default=None, description="Partition coefficient")
|
||
tpsa: Optional[float] = Field(default=None, description="Topological polar surface area")
|
||
melting_point: Optional[float] = Field(default=None, description="In Celsius (°C)")
|
||
|
||
# --- Il valore DAP Calcolato ---
|
||
# Lo impostiamo di default a 0.5 (50%), verrà sovrascritto dal validator
|
||
dap_value: float = 0.5
|
||
|
||
@model_validator(mode='after')
|
||
def compute_dap(self):
|
||
# Lista delle condizioni (True se la condizione riduce l'assorbimento)
|
||
conditions = []
|
||
|
||
# 1. MW > 500 Da
|
||
if self.molecular_weight is not None:
|
||
conditions.append(self.molecular_weight > 500)
|
||
|
||
# 2. High Ionization (Se è True, riduce l'assorbimento)
|
||
if self.high_ionization is not None:
|
||
conditions.append(self.high_ionization is True)
|
||
|
||
# 3. Log Pow <= -1 OR >= 4
|
||
if self.log_pow is not None:
|
||
conditions.append(self.log_pow <= -1 or self.log_pow >= 4)
|
||
|
||
# 4. TPSA > 120 Å2
|
||
if self.tpsa is not None:
|
||
conditions.append(self.tpsa > 120)
|
||
|
||
# 5. Melting Point > 200°C
|
||
if self.melting_point is not None:
|
||
conditions.append(self.melting_point > 200)
|
||
|
||
# LOGICA FINALE:
|
||
# Se c'è almeno una condizione "sicura" (True), il DAP è 0.1
|
||
if any(conditions):
|
||
self.dap_value = 0.1
|
||
else:
|
||
self.dap_value = 0.5
|
||
|
||
return self
|
||
return (DapInfo,)
|
||
|
||
|
||
@app.cell
|
||
def _(DapInfo, dato, re):
|
||
desiderated_keys = ['CAS', 'MolecularWeight', 'XLogP', 'TPSA', 'Melting Point', 'Dissociation Constants']
|
||
actual_keys = [key for key in dato.keys() if key in desiderated_keys]
|
||
|
||
dict = {}
|
||
|
||
for key in actual_keys:
|
||
if key == 'CAS':
|
||
dict['cas'] = dato[key]
|
||
if key == 'MolecularWeight':
|
||
mw = float(dato[key])
|
||
dict['molecular_weight'] = mw
|
||
if key == 'XLogP':
|
||
log_pow = float(dato[key])
|
||
dict['log_pow'] = log_pow
|
||
if key == 'TPSA':
|
||
tpsa = float(dato[key])
|
||
dict['tpsa'] = tpsa
|
||
if key == 'Melting Point':
|
||
try:
|
||
for item in dato[key]:
|
||
if '°C' in item['Value']:
|
||
mp = dato[key]['Value']
|
||
mp_value = re.findall(r"[-+]?\d*\.\d+|\d+", mp)
|
||
if mp_value:
|
||
dict['melting_point'] = float(mp_value[0])
|
||
except:
|
||
continue
|
||
if key == 'Dissociation Constants':
|
||
try:
|
||
for item in dato[key]:
|
||
if 'pKa' in item['Value']:
|
||
pk = dato[key]['Value']
|
||
pk_value = re.findall(r"[-+]?\d*\.\d+|\d+", mp)
|
||
if pk_value:
|
||
dict['high_ionization'] = float(mp_value[0])
|
||
except:
|
||
continue
|
||
|
||
dap_info = DapInfo(**dict)
|
||
dap_info
|
||
return
|
||
|
||
|
||
@app.cell
|
||
def _():
|
||
from pif_compiler.services.srv_cosing import cosing_search, parse_cas_numbers, clean_cosing, identified_ingredients
|
||
return clean_cosing, cosing_search
|
||
|
||
|
||
@app.cell
|
||
def _(clean_cosing, cosing_search):
|
||
raw_cosing = cosing_search("72-48-0", 'cas')
|
||
cleaned_cosing = clean_cosing(raw_cosing)
|
||
cleaned_cosing
|
||
return cleaned_cosing, raw_cosing
|
||
|
||
|
||
@app.cell
|
||
def _(mo):
|
||
mo.md(
|
||
r"""
|
||
otherRestrictions
|
||
refNo
|
||
annexNo
|
||
casNo
|
||
functionName
|
||
"""
|
||
)
|
||
return
|
||
|
||
|
||
@app.cell
|
||
def _(BaseModel, Field, List, Optional):
|
||
class CosingInfo(BaseModel):
|
||
cas : List[str] = Field(default_factory=list)
|
||
common_names : List[str] = Field(default_factory=list)
|
||
inci : List[str] = Field(default_factory=list)
|
||
annex : List[str] = Field(default_factory=list)
|
||
functionName : List[str] = Field(default_factory=list)
|
||
otherRestrictions : List[str] = Field(default_factory=list)
|
||
cosmeticRestriction : Optional[str]
|
||
return (CosingInfo,)
|
||
|
||
|
||
@app.cell
|
||
def _(CosingInfo):
|
||
def cosing_builder(cleaned_cosing):
|
||
cosing_keys = ['nameOfCommonIngredientsGlossary', 'casNo', 'functionName', 'annexNo', 'refNo', 'otherRestrictions', 'cosmeticRestriction', 'inciName']
|
||
keys = [k for k in cleaned_cosing.keys() if k in cosing_keys]
|
||
|
||
cosing_dict = {}
|
||
|
||
for k in keys:
|
||
if k == 'nameOfCommonIngredientsGlossary':
|
||
names = []
|
||
for name in cleaned_cosing[k]:
|
||
names.append(name)
|
||
cosing_dict['common_names'] = names
|
||
if k == 'inciName':
|
||
inci = []
|
||
for inc in cleaned_cosing[k]:
|
||
inci.append(inc)
|
||
cosing_dict['inci'] = names
|
||
if k == 'casNo':
|
||
cas_list = []
|
||
for casNo in cleaned_cosing[k]:
|
||
cas_list.append(casNo)
|
||
cosing_dict['cas'] = cas_list
|
||
if k == 'functionName':
|
||
functions = []
|
||
for func in cleaned_cosing[k]:
|
||
functions.append(func)
|
||
cosing_dict['functionName'] = functions
|
||
if k == 'annexNo':
|
||
annexes = []
|
||
i = 0
|
||
for ann in cleaned_cosing[k]:
|
||
restriction = ann + ' / ' + cleaned_cosing['refNo'][i]
|
||
annexes.append(restriction)
|
||
i = i+1
|
||
cosing_dict['annex'] = annexes
|
||
if k == 'otherRestrictions':
|
||
other_restrictions = []
|
||
for ores in cleaned_cosing[k]:
|
||
other_restrictions.append(ores)
|
||
cosing_dict['otherRestrictions'] = other_restrictions
|
||
if k == 'cosmeticRestriction':
|
||
cosing_dict['cosmeticRestriction'] = cleaned_cosing[k]
|
||
|
||
test_cosing = CosingInfo(
|
||
**cosing_dict
|
||
)
|
||
return test_cosing
|
||
return (cosing_builder,)
|
||
|
||
|
||
@app.cell
|
||
def _(cleaned_cosing, cosing_builder):
|
||
id = cleaned_cosing['identifiedIngredient']
|
||
if id:
|
||
for e in id:
|
||
obj = cosing_builder(e)
|
||
obj
|
||
return
|
||
|
||
|
||
@app.cell
|
||
def _(raw_cosing):
|
||
raw_cosing
|
||
return
|
||
|
||
|
||
@app.cell
|
||
def _():
|
||
return
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app.run()
|