diff --git a/.gitignore b/.gitignore index 23b99e089..045d49b6a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ __pycache__/ bibliovenv/ Bibenv/ -.idea/ \ No newline at end of file +.venv/ +.idea/.DS_Store diff --git a/app.py b/app.py index f0891f894..479f10ceb 100644 --- a/app.py +++ b/app.py @@ -854,8 +854,107 @@ def indicator_types_ui_all(): ), with ui.nav_panel("None", value="API"): - ui.h3("🚧 Warning: API is under construction 🚧") - + ui.h3("πŸ”Œ OpenAlex API", style="color: #5567BB;") + ui.p("Search OpenAlex directly and import the results as a bibliometrix-style dataset.") + + with ui.layout_sidebar(fillable=False, fill=False): + with ui.sidebar(id="sidebar_api", position="right"): + ui.h5("OpenAlex Query", style="color: #5567BB;") + ui.input_text("api_query", "Search query", placeholder="es. machine learning") + ui.input_numeric("api_max_results", "Max results", value=50, min=1, max=200) + ui.input_action_button("start_api_button", "Start", icon=ICONS["play"]) + + @reactive.effect + @reactive.event(input.start_api_button) + def run_api_query(): + # Show loading modal while querying (same style as Historiograph) + def loading_modal(): + phrases = [ + "⏳ Loading... Please wait.", + "πŸ”Ž Querying OpenAlex...", + "πŸ“₯ Downloading records...", + "🧬 Standardizing metadata...", + "πŸ“Š Preparing your dataset...", + "✨ Almost there! Preparing your dashboard...", + ] + modal = ui.modal( + ui.div( + ui.img( + src="https://cisslaboral.laleynext.es/Img/loader-circle.gif", + height="150px", + style="display: block; margin: 0 auto; text-align: center;", + ), + ui.h4( + phrases[0], + id="loading-phrase", + style="font-size: 15px; text-align: center; margin-top: 20px; color: gray;", + ), + ), + easy_close=False, + footer=None, + ) + js = f""" + + """ + return ui.HTML(str(modal) + js) + + ui.modal_show(loading_modal()) + try: + query = (input.api_query() or "").strip() + max_results = int(input.api_max_results()) + + if not query: + ui.notification_show("⚠️ Please enter a search query.", type="warning", duration=5) + return + + # info@bibliometrix.org: indirizzo di progetto gia' usato + # altrove in app.py (sezione About) per la polite pool OpenAlex + result_df, validation_errors = run_openalex_etl( + query, + mailto="info@bibliometrix.org", + max_results=max_results, + ) + + df.set(result_df) + reset_all_analyses() + ui.notification_show( + f"βœ… OpenAlex query completed! The dataset contains {result_df.shape[0]} rows and {result_df.shape[1]} columns.", + duration=5, + close_button=False, + ) + except (ETLPipelineError, OpenAlexRequestError) as e: + ui.notification_show(f"❌ Error querying OpenAlex: {str(e)}", type="error", duration=10) + except Exception as e: + ui.notification_show(f"❌ Unexpected error: {str(e)}", type="error", duration=10) + finally: + ui.modal_remove() + + @render.ui + @reactive.event(input.start_api_button) + def show_api_table(): + if df.get() is None: + return ui.div( + ui.p( + "No dataset loaded yet. Enter a query and click Start.", + style="text-align: center; color: #666; font-size: 16px;" + ), + style="display: flex; flex-direction: column; justify-content: center; align-items: center; height: 150px; border: 2px dashed #ddd; border-radius: 10px; margin: 20px;" + ) + table_ui, _, _ = get_table("OpenAlex", df) + return table_ui + with ui.nav_panel("None", value="collections"): ui.h3("🚧 Warning: Merge Collection is under construction 🚧") @@ -1329,7 +1428,61 @@ async def handle_user_input(user_input: str): answer = "Gemini API key not configured. Please set GEMINI_API_KEY in Settings section." await chat.append_message(answer) - + + # --- Open Access Analysis Section --- + with ui.nav_panel("None", value="open_access_analysis"): + with ui.layout_columns( + col_widths=(9, 3), + style="margin-bottom: -21px;" + ): + with ui.tags.div(style="flex: 1; bottom: 0px;"): + ui.h3("πŸ”“ Open Access Analysis", style="color: #5567BB;") + ui.p("The Open Access status distribution of the dataset") + + with ui.tags.div(style="flex: 2; display: flex; justify-content: flex-end; gap: 5px; align-items: flex-start; bottom: 0px;"): + ui.input_action_button("open_access_report", "Add in Report", icon=ICONS["plus"]) + + todaydate = datetime.today() + todaydate = todaydate.strftime("%Y-%m-%d") + @render.download( + label='πŸ’Ύ Download', + filename=f"OpenAccessAnalysis-{todaydate}.png" + ) + def download_open_access(): + plot_open_access, _ = open_access_informations() + yield plotly_download( + plot_open_access, + title="Open Access Analysis", + height=height.get(), + dpi=dpi.get() + ) + + @render.ui + @reactive.event(input.open_access_report) + def show_open_access_report(): + plots, oa_counts = open_access_informations() + report_excel.set(add_to_report(report_choices, report_excel, [oa_counts], [plots], "openaccessanalysis")) + selection.set(selection.get() + (f"{list(report_choices.get().keys())[-1]}",)) + return ui.notification_show("βœ… Open Access Analysis added to report", duration=5, close_button=False) + + with ui.card(full_screen=True): + @reactive.calc + def open_access_informations(): + return get_open_access_analysis(df) + + with ui.navset_underline(id="open_access_tab"): + with ui.nav_panel("Plot"): + @render_widget + def show_open_access_analysis(): + plot_open_access, oa_counts = open_access_informations() + return plot_open_access + + with ui.nav_panel("Table"): + @render.ui + def table_open_access_analysis(): + _, oa_counts = open_access_informations() + return ui.HTML(DT(oa_counts, style="width=100%;")) + # --- Average Citations per Year Section --- with ui.nav_panel("None", value="average_citations_per_year"): with ui.layout_columns( @@ -8185,7 +8338,7 @@ def update_plot_settings(): # --- Sidebar Management --- @render.express() -@reactive.event(input.start_button) +@reactive.event(input.start_button, input.start_api_button) def toggle_sidebar(): with ui.tags.div(id="sidebar_2", class_="custom-sidebar"): with ui.accordion(id="sidebar_accordion_data", multiple=False, open=False): @@ -8206,6 +8359,7 @@ def toggle_sidebar(): with ui.accordion_panel("Overview", icon=ICONS["play_colored"]): ui.input_action_button("go_main", "Main Information", class_="sidebar-button", icon=ICONS["overview"]) ui.input_action_button("go_annual_scientific_production", "Annual Scientific Production", class_="sidebar-button", icon=ICONS["annual_growth_rate"]) + ui.input_action_button("go_open_access_analysis", "Open Access Analysis", class_="sidebar-button", icon=ICONS["sources"]) ui.input_action_button("go_average_citations_per_year", "Average Citations per Year", class_="sidebar-button", icon=ICONS["average_citations_per_doc"]) ui.input_action_button("go_three_field_plot", "Three-Field Plot", class_="sidebar-button", icon=ICONS["overview"]) with ui.accordion_panel("Sources", icon=ICONS["sources_colored"]): @@ -8422,6 +8576,11 @@ def _(): def _(): ui.update_navs("hidden_tabs", selected="annual_scientific_production") +@reactive.effect +@reactive.event(input.go_open_access_analysis) +def _(): + ui.update_navs("hidden_tabs", selected="open_access_analysis") + @reactive.effect @reactive.event(input.go_average_citations_per_year) def _(): diff --git a/functions/__init__.py b/functions/__init__.py index 20e24de36..3ce60d80e 100644 --- a/functions/__init__.py +++ b/functions/__init__.py @@ -20,6 +20,7 @@ from .get_localcitedsources import * from .get_lotkalaw import * from .get_maininformations import * +from .get_openaccessanalysis import * from .get_referencesspectroscopy import * from .get_relevantaffiliations import * from .get_relevantauthors import * diff --git a/functions/get_clusteringcoupling.py b/functions/get_clusteringcoupling.py index 8263a46b3..de7517027 100644 --- a/functions/get_clusteringcoupling.py +++ b/functions/get_clusteringcoupling.py @@ -1,11 +1,24 @@ from www.services import * -def get_clustering_coupling(df, unit_of_analysis, coupling_measured, stemmer, impact_measure, - cluster_labeling, ngram, num_of_units, min_cluster_freq, - label_per_cluster, label_size, community_repulsion, +def get_clustering_coupling(df, unit_of_analysis, coupling_measured, stemmer, impact_measure, + cluster_labeling, ngram, num_of_units, min_cluster_freq, + label_per_cluster, label_size, community_repulsion, clustering_algorithm, node_shape='dot'): - + # LIMITE NOTO (non investigato oltre in questa sessione, fuori scope): + # "Cluster by Coupling" risulta lento con dati OpenAlex per una causa non + # identificata nella costruzione della matrice di coupling/coincidenza + # citazionale (couplingMap -> network -> biblionetwork/cocMatrix/ + # network_plot in www/services/couplingmap.py), indipendente dal nostro + # standardizzatore. La dimensione del calcolo principale (matrice N x N + # documenti, N ~ 30) e' teoricamente troppo piccola per giustificare una + # lentezza reale, quindi non sembra un limite "normale ma lento" per + # questo volume di dati - ma la causa esatta non e' stata isolata (analisi + # non validata, si sospetta codice preesistente del professore, non + # necessariamente legato al formato di CR). Separato dal bug gia' corretto + # in couplingmap.py::localCitations (histNetwork che ritornava None senza + # gestione), che riguarda un crash successivo, non questa lentezza. + # Generate coupling map coupling_map = couplingMap( df, diff --git a/functions/get_data.py b/functions/get_data.py index 16baed992..7fea66374 100644 --- a/functions/get_data.py +++ b/functions/get_data.py @@ -1,6 +1,34 @@ from www.services import * +def _split_multivalue_cell(value): + """ + Ricostruisce una list[str] a partire da una cella Excel contenente una + rappresentazione "; "-separata (o ";"-separata, senza spazio) di una + colonna multi-valore (vedi type_contracts.py::COLUMN_SPECS, STRING_LIST). + + Gestisce robustamente i casi limite prodotti da un roundtrip Excel: + - cella vuota/NaN (tipico di pd.read_excel su una cella Excel vuota, + che arriva come float NaN, non come stringa vuota) -> lista vuota; + - stringa vuota o solo spazi -> lista vuota; + - elementi vuoti generati da separatori ripetuti/spazi spuri -> scartati. + + Args: + value: contenuto grezzo della cella cosi' come restituito da + pd.read_excel (str, float NaN, o altro tipo scalare). + + Returns: + list[str]: elementi ricostruiti, lista vuota se value non contiene + dati utilizzabili. + """ + if pd.isna(value): + return [] + text = str(value).strip() + if not text: + return [] + return [item.strip() for item in re.split(r";\s*", text) if item.strip()] + + def get_data(input, database, df, reset_callback=None): """ Handle the data upload and display process. @@ -67,7 +95,32 @@ def get_data(input, database, df, reset_callback=None): ) elif input.select() == "1B": - df.set(pd.read_excel(file[0]["datapath"])) + loaded = pd.read_excel(file[0]["datapath"]) + + # DEBUGGING LOG (secondo esempio di "Poor handling of missing values" + # nel codice originale, oltre al bug PY str-vs-int): i file .xlsx non + # possono contenere oggetti Python nativi. Un DataFrame con colonne + # multi-valore (AU, AF, C1, CR, DE, ID, AU_UN - list[str], vedi + # type_contracts.py::COLUMN_SPECS) scritto con df.to_excel() viene + # serializzato da pandas con str(list) per cella (es. + # "['Autore1', 'Autore2']"), e pd.read_excel() lo rilegge cosi' com'e': + # una stringa contenente il repr letterale della lista, non una lista + # vera. Verificato concretamente: senza questa conversione, + # functions/get_relevant_authors.py va in + # `ValueError: cannot convert float NaN to integer` perche' AU non e' + # piu' esplodibile come lista (get_relevant_authors.py:108). Qui + # ricostruiamo le liste dalla rappresentazione "; "-separata che il + # nostro export di test scrive al posto del repr Python (vedi + # standard del brief: multi-valore = stringa unica joinata con "; "). + multivalue_columns = [ + column for column, spec in COLUMN_SPECS.items() + if spec["type"] == ColumnType.STRING_LIST + ] + for column in multivalue_columns: + if column in loaded.columns: + loaded[column] = loaded[column].apply(_split_multivalue_cell) + + df.set(loaded) # Reset all analysis results when new dataset is loaded if reset_callback: reset_callback() diff --git a/functions/get_historiograph.py b/functions/get_historiograph.py index 089d02387..8d9bd24e2 100644 --- a/functions/get_historiograph.py +++ b/functions/get_historiograph.py @@ -25,11 +25,45 @@ def get_historiograph(df, node_label="AU1", histNodes=20, hist_isolates=True, hi hist_plot: oggetto con layout e grafo networkx hist_data: dataframe con metadati, DOI cliccabili, cluster, anni filename: nome del file HTML interattivo salvato temporaneamente + + None se la sorgente del DataFrame non e' supportata per l'analisi + citazionale diretta (vedi nota sotto), invece di sollevare TypeError. """ # Pre-elaborazione df = metaTagExtraction(df, "SR") hist_results = histNetwork(df, min_citations=0, sep=sep, network=True) + # LIMITE NOTO (debugging step, vedi anche get_local_cited_authors.py e + # get_local_cited_documents.py per lo stesso pattern): www/services/histnetwork.py::histNetwork + # supporta solo DB == "Web_of_Science" o "Scopus" (righe 37-43 di quel file); + # per qualunque altro valore di DB β€” incluso il nostro "OPENALEX", ma anche + # Dimensions/The_Lens/PubMed/Cochrane della pipeline storica stessa β€” stampa + # "Database not compatible with direct citation analysis" e ritorna None + # PRIMA di toccare la colonna CR. Non e' quindi un problema di formato di CR: + # la funzione non arriva mai a parsarlo per queste sorgenti. + # + # Deliberatamente NON abbiamo esteso histNetwork con un ramo "OPENALEX": + # 1) il ramo wos() dipende da M['SR_FULL'], colonna che la nostra pipeline + # scarta deliberatamente perche' non fa parte dello schema canonico a 34 + # colonne (vedi openalex_mapper.py::_compute_calculated_fields) β€” andrebbe + # comunque in KeyError; + # 2) anche risolvendo (1), wos() cerca auto-citazioni ESATTE all'interno + # della stessa collezione (un paper che cita un altro paper anch'esso nei + # risultati): su un campione generico di poche decine di risultati da una + # query testuale, la rete risultante sarebbe quasi sempre vuota β€” non un + # bug, ma un risultato di scarso valore che non giustifica lo sforzo; + # 3) il ramo scopus() si aspetta l'anno tra parentesi nel riferimento + # (regex r'.*\((\d{4})\).*'), formato diverso dal nostro CR + # "Autore, ANNO, Rivista" (anno non tra parentesi) β€” anche qui non un + # crash, ma zero citazioni valide trovate silenziosamente. + # Qui ci limitiamo quindi a intercettare il None e propagarlo pulito, con lo + # stesso significato di "nessun risultato" gia' usato ovunque in app.py per + # questo tipo di analisi (historiograph_results = reactive.Value(None), con + # ogni consumer che fa `if result is not None` e mostra un placeholder + # altrimenti) β€” nessuna nuova convenzione introdotta. + if hist_results is None: + return None + # 1. Costruzione iniziale del grafo hist_plot = histPlot( hist_results, diff --git a/functions/get_localcitedauthors.py b/functions/get_localcitedauthors.py index e663192bc..08eb5efd6 100644 --- a/functions/get_localcitedauthors.py +++ b/functions/get_localcitedauthors.py @@ -12,7 +12,13 @@ def get_local_cited_authors(df, num_of_cited_authors, fast_search=False): Returns: A Plotly figure object and a DataFrame of the most local cited authors. - """ + + None (non una tupla) se la sorgente del DataFrame non e' supportata + per l'analisi citazionale diretta (vedi nota sotto), invece di + sollevare TypeError β€” stesso singolo valore sentinella che app.py + verifica con `if result is None` al punto di chiamata + (local_cited_authors_result.set(result) senza spacchettare prima). + """ # Determine the local citation threshold if fast_search: loccit = df['TC'].quantile(0.75) @@ -21,12 +27,25 @@ def get_local_cited_authors(df, num_of_cited_authors, fast_search=False): df = metaTagExtraction(df, "SR") M = df.get() - + # Fill missing values M['TC'] = M['TC'].fillna(0) # Create a histogram network H = histNetwork(df, min_citations=loccit, sep=";", network=False) + + # LIMITE NOTO (debugging step, spiegazione completa in + # get_historiograph.py): histNetwork ritorna None per qualunque DB diverso + # da "Web_of_Science"/"Scopus" (incluso il nostro "OPENALEX"), PRIMA di + # toccare CR β€” non e' un problema di formato dei riferimenti. Non estendiamo + # histNetwork qui (richiederebbe M['SR_FULL'], che scartiamo deliberatamente, + # e produrrebbe comunque reti quasi sempre vuote su campioni tipici). + # Propaghiamo (None, None) invece di un TypeError: stesso sentinel None gia' + # gestito da ogni consumer in app.py (local_cited_authors_result = + # reactive.value(None), con `if result is None` -> placeholder). + if H is None: + return None + LCS = H['histData'] M = H['M'] diff --git a/functions/get_localciteddocuments.py b/functions/get_localciteddocuments.py index 1dea8d5a5..5a1ccfa31 100644 --- a/functions/get_localciteddocuments.py +++ b/functions/get_localciteddocuments.py @@ -12,6 +12,12 @@ def get_local_cited_documents(df, num_of_local_cited_docs, field_separator, fast Returns: A Plotly figure object and a DataFrame of the most local cited documents. + + None (non una tupla) se la sorgente del DataFrame non e' supportata + per l'analisi citazionale diretta (vedi nota sotto), invece di + sollevare TypeError β€” stesso singolo valore sentinella che app.py + verifica con `if result is None` al punto di chiamata + (local_cited_documents_results.set(result) senza spacchettare prima). """ df = metaTagExtraction(df, "SR") M = df.get() @@ -21,12 +27,22 @@ def get_local_cited_documents(df, num_of_local_cited_docs, field_separator, fast loccit = M['TC'].quantile(0.75) else: loccit = 1 - + # Fill missing values M['TC'] = M['TC'].fillna(0) # Create a histogram network H = histNetwork(df, min_citations=loccit, sep=";", network=False) + + # LIMITE NOTO (debugging step, spiegazione completa in + # get_historiograph.py): histNetwork ritorna None per qualunque DB diverso + # da "Web_of_Science"/"Scopus" (incluso il nostro "OPENALEX"), PRIMA di + # toccare CR β€” non e' un problema di formato dei riferimenti. Non estendiamo + # histNetwork qui (richiederebbe M['SR_FULL'], che scartiamo deliberatamente, + # e produrrebbe comunque reti quasi sempre vuote su campioni tipici). + if H is None: + return None + LCS = H['histData'] M = H['M'] diff --git a/functions/get_openaccessanalysis.py b/functions/get_openaccessanalysis.py new file mode 100644 index 000000000..0f47786d8 --- /dev/null +++ b/functions/get_openaccessanalysis.py @@ -0,0 +1,65 @@ +from www.services import * + + +def get_open_access_analysis(df): + """ + Generate a pie chart and table of the Open Access status distribution. + + Args: + df: A DataFrame object containing the data. + + Returns: + A Plotly figure object representing the Open Access distribution and + a DataFrame with the document count per OA status. + """ + data = df.get() + + # OA vuoto ("") indica che lo stato Open Access non e' noto/disponibile per + # quel documento (comune sia nella pipeline OpenAlex sia nelle altre fonti + # storiche quando il dato non e' popolato). Scelta: invece di scartare questi + # record dal conteggio (il che nasconderebbe quanto e' effettivamente + # completo il dataset), li raggruppiamo in una categoria esplicita + # "Unknown" - coerente con lo spirito delle tabelle di completezza gia' + # presenti altrove nell'app (vedi functions/get_table.py). + oa_status = data["OA"].fillna("").replace("", "Unknown") + oa_counts = oa_status.value_counts().reset_index() + oa_counts.columns = ["OA Status", "Freq"] + oa_counts = oa_counts.sort_values(by="Freq", ascending=False).reset_index(drop=True) + + # Create the plot + fig = px.pie( + oa_counts, names="OA Status", values="Freq", + hole=0.4, + color_discrete_sequence=px.colors.sequential.Blues_r, + ) + + # Customize the layout and tooltips (hover) + fig.update_traces( + textinfo="label+percent", + textfont=dict(size=13), + marker=dict(line=dict(color="white", width=2)), + hovertemplate=( + "%{label}
" + "Documents: %{value}" + ) + ) + + fig.update_layout( + plot_bgcolor='white', + font=dict(color="#222222", size=14, family="Segoe UI, Arial"), + margin=dict(l=50, r=30, t=60, b=50), + height=600, + legend=dict(orientation='h', yanchor='bottom', y=-0.15, xanchor='center', x=0.5), + hoverlabel=dict( + bgcolor="white", + font_size=13, + font_family="Segoe UI, Arial", + bordercolor="#1f77b4" + ) + ) + + fig = go.FigureWidget(fig) + fig._config = fig._config | {'modeBarButtonsToRemove': ['pan', 'select', 'lasso2d', 'toImage'], + 'displaylogo': False} + + return fig, oa_counts diff --git a/www/services/__init__.py b/www/services/__init__.py index 28584e105..8b6b3fefe 100644 --- a/www/services/__init__.py +++ b/www/services/__init__.py @@ -1,6 +1,7 @@ from .biblionetwork import * from .cocmatrix import * from .couplingmap import * +from .etl_pipeline import * from .format_functions import * from .histnetwork import * from .histplot import * @@ -8,10 +9,13 @@ from .igraph2vis import * from .metatagextraction import * from .networkplot import * +from .openalex_client import * +from .openalex_mapper import * from .parsers import * from .plotlydownload import * from .savereport import * from .tabletag import * from .termextraction import * from .thematicmap import * +from .type_contracts import * from .utils import * \ No newline at end of file diff --git a/www/services/couplingmap.py b/www/services/couplingmap.py index a2b3628d7..3887948e1 100644 --- a/www/services/couplingmap.py +++ b/www/services/couplingmap.py @@ -525,6 +525,27 @@ def localCitations(df, fast_search=False, sep=";"): loccit = 1 H = histNetwork(df, min_citations=loccit, sep=sep, network=False) + + # LIMITE NOTO (stesso pattern gia' applicato in get_historiograph.py, + # get_localcitedauthors.py, get_localciteddocuments.py): histNetwork ritorna + # None per qualunque DB diverso da "Web_of_Science"/"Scopus" (quindi anche + # per "OPENALEX"), PRIMA di calcolare alcunche' - non e' un problema di + # formato di CR, e per le stesse ragioni gia' documentate in + # get_historiograph.py non estendiamo histNetwork qui. A differenza di quei + # tre file pero', qui il chiamante (normalizeCitationScore, quindi + # couplingMap/get_clustering_coupling.py) si aspetta sempre un dict con una + # colonna 'LCS' popolata in 'M': costruiamo un fallback con LCS=0 per ogni + # documento (nessun dato di citazione locale disponibile per questa + # sorgente) invece di propagare un TypeError su H['histData']. + if H is None: + M_fallback = M.copy() + M_fallback['LCS'] = 0 + return { + 'Authors': pd.DataFrame(columns=["Authors", "N. of Local Citations"]), + 'Papers': pd.DataFrame(columns=["Paper", "DOI", "Year", "LCS", "GCS"]), + 'M': M_fallback, + } + LCS = H['histData'] M = H['M'] diff --git a/www/services/etl_pipeline.py b/www/services/etl_pipeline.py new file mode 100644 index 000000000..7d99dfcfd --- /dev/null +++ b/www/services/etl_pipeline.py @@ -0,0 +1,366 @@ +""" +Entry-point unico per la pipeline ETL "query utente -> OpenAlex -> schema WoS-style". + +Orchestra, in ordine: +1. query utente -> openalex_client (ricerca + paginazione cursor-based, con + risoluzione opzionale in batch degli ID in `referenced_works`) +2. openalex_client -> openalex_mapper (mapping di ciascun "work" grezzo sulle + 34 colonne dello schema bibliometrix-style) +3. calcolo dei campi derivati che richiedono l'intera collezione (es. SR, che deve + deduplicare le sigle "Autore, Anno, Rivista" ripetute nel dataset, con la stessa + logica concettuale di metatagextraction.py::SR) +4. type_contracts (coercizione e validazione dei tipi prima di finalizzare l'output) +5. assemblaggio del pandas.DataFrame finale, con lo stesso schema a 34 colonne + prodotto dalla pipeline storica basata su file (vedi + www/services/format_functions.py::process_single_file). + +Questo modulo e' pensato come punto di ingresso alternativo a +functions/get_data.py (che copre l'import da file WoS/Scopus/ecc.), cosi' che il +resto della codebase (le funzioni in functions/get_*.py, che operano sul +DataFrame condiviso `df`) possa continuare a funzionare senza sapere se i dati +provengono da un file caricato dall'utente o da una query OpenAlex. +""" + +from .utils import * +from .openalex_client import * +from .openalex_mapper import * +from .type_contracts import * + + +# Budget di default (secondi) per l'intera fase di risoluzione batch dei +# referenced_works (vedi _resolve_references_for_works). Esposto anche come +# parametro pubblico di run_openalex_etl, cosi' e' configurabile senza +# toccare il codice interno. +DEFAULT_RESOLVE_TIMEOUT_SECONDS = 30 + + +class ETLPipelineError(Exception): + """Sollevata quando la pipeline ETL fallisce in uno dei suoi stadi (fetch, + risoluzione referenze, mapping, validazione) in un modo che non permette di + produrre un DataFrame utilizzabile.""" + pass + + +def run_openalex_etl( + query, + filters=None, + mailto=None, + max_results=None, + resolve_references=True, + strict_validation=True, + resolve_timeout_seconds=DEFAULT_RESOLVE_TIMEOUT_SECONDS, + fetch_timeout_seconds=DEFAULT_FETCH_TIMEOUT_SECONDS, +): + """ + Entry-point principale: esegue l'intera pipeline ETL da una query utente + OpenAlex a un pandas.DataFrame nello schema a 34 colonne WoS-style. + + Args: + query: stringa di ricerca testuale libera, inoltrata a + openalex_client.search_works. + filters: dict opzionale di filtri OpenAlex aggiuntivi (es. anno, tipo + documento), vedi openalex_client.search_works. + mailto: email da usare per la polite pool di OpenAlex. + max_results: numero massimo di record da scaricare; None per scaricare + tutti i risultati della query. + resolve_references: se True, risolve in batch gli ID in `referenced_works` + per popolare la colonna CR con citazioni leggibili invece dei soli ID + OpenAlex (costo aggiuntivo in chiamate HTTP, vedi + openalex_client.get_works_by_ids). + fetch_timeout_seconds: budget di tempo (secondi) per l'INTERA + paginazione della ricerca iniziale (non per singola richiesta HTTP), + passato a openalex_client.search_works tramite _fetch_raw_works. + Default DEFAULT_FETCH_TIMEOUT_SECONDS (30s): oltre questo tetto, le + pagine non ancora richieste vengono saltate (nessuna eccezione) e + si procede con i work gia' raccolti fino a quel momento. None per + nessun limite (comportamento pre-esistente, sconsigliato: e' la + fase in cui e' stato diagnosticato un blocco reale di oltre due + minuti senza mai un'eccezione). + resolve_timeout_seconds: budget di tempo (secondi) per l'INTERA fase di + risoluzione dei riferimenti (non per singola chiamata HTTP), passato + a _resolve_references_for_works. Default DEFAULT_RESOLVE_TIMEOUT_SECONDS + (30s): oltre questo tetto, i batch non ancora processati vengono + saltati (nessuna eccezione) e i riferimenti corrispondenti ricadono + sul fallback a ID nudo in format_cr_column, invece di bloccare + l'intera pipeline su query con molte referenze da risolvere. + None per nessun limite (comportamento pre-esistente, sconsigliato + su query generiche/con molti risultati). Ignorato se + resolve_references=False. + strict_validation: passato come `strict` a type_contracts.validate_record + per ogni record dopo la coercizione dei tipi: se True (default), + eventuali colonne non riconosciute nello schema canonico contano come + errori di validazione; se False, vengono tollerate. In ENTRAMBI i + casi, qualunque altro errore residuo (tipo non conforme, colonna + obbligatoria mancante, None/NaN sopravvissuto alla coercizione) fa + comunque sollevare ETLPipelineError da _validate_records: e' una rete + di sicurezza interna che non dovrebbe mai scattare in condizioni + normali (map_work_to_record + coerce_record_types garantiscono gia' + record conformi), non un comportamento disattivabile con questo flag. + + Returns: + Tupla (df, validation_errors): + - df: pandas.DataFrame con lo schema a 34 colonne bibliometrix-style, + colonne nell'ordine di `columns` (www/services/utils.py). + - validation_errors: list[str], sempre [] nel percorso di successo + (qualunque errore di validazione residuo interrompe la pipeline con + ETLPipelineError prima di raggiungere il return β€” vedi + _validate_records). Il secondo elemento della tupla e' mantenuto per + stabilita' della firma pubblica, per il caso in cui in futuro + _validate_records venga reso tollerante invece che bloccante. + + Raises: + ETLPipelineError: se uno stadio non recuperabile della pipeline fallisce + (nessun risultato dalla query, errore HTTP non gestito dal client, + oppure errori di validazione residui dopo la coercizione dei tipi). + """ + works = _fetch_raw_works(query, filters, mailto, max_results, fetch_timeout_seconds=fetch_timeout_seconds) + + resolved_references = None + if resolve_references: + resolved_references = _resolve_references_for_works( + works, mailto, resolve_timeout_seconds=resolve_timeout_seconds + ) + + records = _map_works_to_records(works, resolved_references=resolved_references) + records = _compute_calculated_fields(records) + + coerced_records, _ = _validate_records(records, strict=strict_validation) + + df = _build_dataframe(coerced_records) + + return df, [] + + +def _fetch_raw_works(query, filters, mailto, max_results, fetch_timeout_seconds=DEFAULT_FETCH_TIMEOUT_SECONDS): + """ + Stadio 1: recupera dalla API OpenAlex la lista grezza di oggetti "work" (dict + JSON) corrispondenti alla query utente, delegando a + openalex_client.search_works (che gia' pagina internamente con cursore fino + a max_results o esaurimento dei risultati). + + LIMITE DI SICUREZZA (debugging log): search_works riceve qui il budget di + tempo `fetch_timeout_seconds` (vedi il suo docstring per il dettaglio del + caso reale diagnosticato: una singola richiesta rimasta bloccata >2m44s + senza mai sollevare un'eccezione di timeout). Se il budget scade a meta' + paginazione, search_works restituisce i risultati raccolti fino a quel + momento invece di bloccare - _fetch_raw_works non tratta questo come un + errore: una lista parziale ma non vuota e' comunque un risultato valido + per il resto della pipeline. + + Args: + query, filters, mailto, max_results: vedi run_openalex_etl. + fetch_timeout_seconds: budget di tempo (secondi) per l'INTERA + paginazione di search_works. Default DEFAULT_FETCH_TIMEOUT_SECONDS + (30s, definito in openalex_client.py). None per nessun limite. + + Returns: + list[dict]: oggetti "work" OpenAlex grezzi (eventualmente parziali se + il budget di tempo e' stato superato durante la paginazione). + + Raises: + ETLPipelineError: se la query non produce alcun risultato, oppure se + openalex_client.search_works solleva OpenAlexRequestError (errore + HTTP non recuperabile dopo i retry). + """ + try: + works = search_works( + query, + max_results=max_results, + filters=filters, + mailto=mailto, + fetch_timeout_seconds=fetch_timeout_seconds, + ) + except OpenAlexRequestError as exc: + raise ETLPipelineError( + f"Recupero dei work da OpenAlex fallito per la query {query!r}: {exc}" + ) from exc + + if not works: + raise ETLPipelineError(f"Nessun risultato OpenAlex per la query {query!r}.") + + return works + + +def _resolve_references_for_works(works, mailto, resolve_timeout_seconds=DEFAULT_RESOLVE_TIMEOUT_SECONDS): + """ + Stadio 2 (opzionale): raccoglie l'unione di tutti gli ID presenti nel campo + `referenced_works` dei work scaricati e li risolve in batch tramite + openalex_client.get_works_by_ids, per costruire citazioni leggibili per la + colonna CR invece dei soli ID. + + Deduplica a livello di INTERA collezione (non per singolo work) e chiama + get_works_by_ids UNA SOLA VOLTA sull'unione di tutti gli ID: get_works_by_ids + spezza gia' internamente in chunk da 50 (il suo DEFAULT_BATCH_SIZE), quindi + minimizzare qui il numero di chiamate a get_works_by_ids stessa (una sola, + con l'intera lista) minimizza a cascata il numero di richieste HTTP totali + rispetto a chiamarla una volta per work. + + Ogni referenced_works di un singolo work viene troncato a + openalex_mapper.MAX_REFERENCED_WORKS PRIMA di entrare nell'unione: sono gli + stessi ID che format_cr_column considerera' comunque (lo stesso cap), quindi + risolvere ID oltre quel limite sarebbe lavoro sprecato. + + LIMITE DI SICUREZZA (debugging log): una query generica con molti risultati + puo' generare migliaia di ID da risolvere (es. 30 work x MAX_REFERENCED_WORKS=100 + = fino a 3000 ID, cioe' 60 batch da 50). Senza un tetto, nel caso peggiore di + errori di rete ripetuti su ogni batch, il tempo totale poteva arrivare a + decine di minuti (~45 min con i retry di default), bloccando l'intero + handler Shiny sincrono che chiama run_openalex_etl. Il cronometro parte QUI, + all'inizio di questa funzione (time.monotonic()), e viene passato a + get_works_by_ids, che lo controlla prima di iniziare ogni nuovo batch: se il + budget e' superato, i batch rimanenti vengono saltati (nessuna eccezione) e + si restituisce il dizionario parziale gia' risolto. I riferimenti non + risolti in tempo ricadono sul fallback a ID nudo gia' esistente in + openalex_mapper.py::format_cr_column - degradazione, non blocco. + + Args: + works: list[dict] di work OpenAlex grezzi, vedi _fetch_raw_works. + mailto: email per la polite pool. + resolve_timeout_seconds: budget di tempo (secondi) per l'INTERA + risoluzione (non per singolo batch). Default + DEFAULT_RESOLVE_TIMEOUT_SECONDS (30s). None per nessun limite. + + Returns: + dict[str, dict]: mappa da ID OpenAlex a oggetto "work" risolto, passata a + openalex_mapper.map_work_to_record / format_cr_column. Dizionario vuoto + se nessun work ha referenced_works. Puo' essere parziale se il budget di + tempo e' stato superato prima di processare tutti i batch. + """ + start_time = time.monotonic() + + seen = set() + all_ids = [] + for work in works: + referenced = (work.get("referenced_works") or [])[:MAX_REFERENCED_WORKS] + for ref_id in referenced: + if ref_id and ref_id not in seen: + seen.add(ref_id) + all_ids.append(ref_id) + + if not all_ids: + return {} + + return get_works_by_ids( + all_ids, + mailto=mailto, + resolve_timeout_seconds=resolve_timeout_seconds, + start_time=start_time, + ) + + +def _map_works_to_records(works, resolved_references=None): + """ + Stadio 3: applica openalex_mapper.map_work_to_record a ciascun work grezzo, + producendo la lista di record (dict) nello schema a 34 colonne WoS-style + (esclusi i campi calcolati a livello di collezione come SR). + + Args: + works: list[dict] di work OpenAlex grezzi. + resolved_references: dict opzionale id -> work risolto, vedi + _resolve_references_for_works; None se resolve_references=False in + run_openalex_etl (CR contera' presumibilmente solo gli ID grezzi). + + Returns: + list[dict]: un record per work, con le chiavi delle 34 colonne (tranne i + campi calcolati a livello di collezione). + """ + return [ + map_work_to_record(work, resolved_references=resolved_references) + for work in works + ] + + +def _compute_calculated_fields(records): + """ + Stadio 4: calcola i campi derivati dall'intera collezione, in particolare SR + (Autore, Anno, Rivista), che richiede la deduplicazione tra tutti i record del + dataset (stessa logica concettuale di metatagextraction.py::SR, adattata per + operare su una lista di record invece che su un DataFrame reattivo wrappato da + `df.get()`/`df.set()`). + + Args: + records: list[dict] prodotta da _map_works_to_records. + + Returns: + list[dict]: nuovi record (gli originali non vengono mutati, vedi + compute_sr_for_records), arricchiti con la chiave "SR". Nota: la + funzione riusata, openalex_mapper.compute_sr_for_records, aggiunge + anche "SR_FULL" (sottoprodotto di metatagextraction.py::SR, riusata + invariata) β€” SR_FULL viene deliberatamente SCARTATA qui perche' non fa + parte dello schema canonico a 34 colonne (`columns` in utils.py): + decisione presa esplicitamente dopo averlo verificato con + type_contracts.validate_record durante lo sviluppo di questo modulo. + """ + enriched = compute_sr_for_records(records) + for record in enriched: + record.pop("SR_FULL", None) + return enriched + + +def _validate_records(records, strict=True): + """ + Stadi 5+6: prima coercizione (type_contracts.coerce_record_types su ogni + record, per assorbire inconsistenze di tipo note es. TC/PY come stringa), + poi validazione (type_contracts.validate_record sui record gia' coerciti). + + Se, DOPO la coercizione, restano errori di validazione, solleva + ETLPipelineError con il dettaglio invece di restituirli: coerce_record_types + e' pensata per garantire sempre output conforme, quindi un errore residuo a + questo punto significa una regressione nella pipeline stessa (es. un + format_XX_column che ha smesso di rispettare il proprio contratto di tipo), + non un problema recuperabile sui dati di un singolo work β€” e' la "rete di + sicurezza finale" del brief, non un controllo disattivabile. + + Args: + records: list[dict] da validare, dopo il calcolo dei campi derivati + (vedi _compute_calculated_fields). + strict: propagato a type_contracts.validate_record; se True, record con + colonne non riconosciute sono considerati invalidi. + + Returns: + Tupla (coerced_records, errors): + - coerced_records: list[dict] con i valori coerciti da + type_contracts.coerce_record_types. + - errors: list[str], SEMPRE [] quando la funzione ritorna normalmente + (qualunque errore residuo fa sollevare ETLPipelineError prima del + return). Mantenuta nella firma per stabilita' dell'API. + + Raises: + ETLPipelineError: se validate_record rileva almeno un errore su almeno + un record dopo la coercizione. + """ + coerced_records = [coerce_record_types(record) for record in records] + + errors = [] + for index, record in enumerate(coerced_records): + for error in validate_record(record, strict=strict): + errors.append(f"record {index}: {error}") + + if errors: + raise ETLPipelineError( + "Validazione fallita su record gia' coerciti (non dovrebbe accadere): " + + "; ".join(errors) + ) + + return coerced_records, errors + + +def _build_dataframe(records): + """ + Stadio 6: assembla il pandas.DataFrame finale a partire dalla lista di record + validati, garantendo che le colonne siano nello stesso ordine dello schema a + 34 colonne definito in www/services/utils.py (variabile `columns`), per + compatibilita' con le funzioni esistenti in functions/get_*.py che assumono + questo schema. + + Args: + records: list[dict] di record coerciti/validati. + + Returns: + pandas.DataFrame con lo schema a 34 colonne WoS-style, colonne ordinate + secondo `columns` (www/services/utils.py). Con records=[] restituisce un + DataFrame a 0 righe ma con tutte le 34 colonne comunque definite. + """ + df = pd.DataFrame(records) + df = df.reindex(columns=columns) + return df diff --git a/www/services/openalex_client.py b/www/services/openalex_client.py new file mode 100644 index 000000000..0a72e92ae --- /dev/null +++ b/www/services/openalex_client.py @@ -0,0 +1,457 @@ +""" +Client HTTP per l'API pubblica di OpenAlex (https://api.openalex.org). + +Responsabilita' di questo modulo: +- Eseguire ricerche testuali sull'endpoint /works. +- Gestire la paginazione cursor-based per scaricare risultati oltre la singola pagina. +- Applicare retry con backoff esponenziale sugli errori transitori (rate limit 429, + errori 5xx, timeout di rete). +- Risolvere in batch una lista di ID OpenAlex (usato per popolare la colonna CR a + partire dagli ID grezzi contenuti in `referenced_works`). + +Questo modulo non fa alcun mapping verso lo schema WoS-style: restituisce sempre i +dizionari JSON grezzi cosi' come li restituisce OpenAlex. Il mapping verso le 34 +colonne e' responsabilita' di openalex_mapper.py; l'orchestrazione dei due e' +responsabilita' di etl_pipeline.py. +""" + +import logging + +from .utils import * + + +logger = logging.getLogger(__name__) + + +BASE_URL = "https://api.openalex.org" +WORKS_ENDPOINT = f"{BASE_URL}/works" + +DEFAULT_PER_PAGE = 25 +MAX_PER_PAGE = 200 # limite massimo imposto dalle API OpenAlex +DEFAULT_MAX_RETRIES = 3 +DEFAULT_BACKOFF_FACTOR = 1.5 + +# Tupla (connect_timeout, read_timeout), non un singolo valore. DEBUGGING LOG: +# con un timeout singolo (era 10s), `requests` lo applica come timeout di +# INATTIVITA' tra un chunk di risposta e il successivo, NON come tetto sul +# tempo totale della richiesta - se il server (o un proxy/CDN intermedio) +# manda anche un solo byte ogni tanto entro la finestra, la richiesta puo' +# restare appesa indefinitamente senza mai sollevare ReadTimeout. Diagnosticato +# concretamente: una query "AI" e' rimasta bloccata >2m44s su una singola +# chiamata di search_works (connessione TCP ESTABLISHED verso l'infrastruttura +# OpenAlex, CPU 0%, nessun retry mai scattato) prima di essere interrotta +# manualmente. connect_timeout=5s (tempo per stabilire la connessione TCP), +# read_timeout=15s (silenzio massimo tollerato tra un byte e l'altro della +# risposta) restano lo stesso tipo di garanzia "anti-inattivita'", ma con +# margini piu' stretti; il vero argine contro un'attesa indefinita e' pero' +# il tetto di tempo complessivo aggiunto a search_works (vedi piu' sotto) e a +# get_works_by_ids, che non dipende da come si comporta il timeout di requests. +DEFAULT_TIMEOUT = (5, 15) + +DEFAULT_BATCH_SIZE = 50 # limite OpenAlex per filter=openalex_id:ID1|ID2|... + +# max_retries ridotto, isolato a get_works_by_ids (risoluzione batch di +# referenced_works per popolare CR). NON tocca DEFAULT_MAX_RETRIES, che resta +# a 3 per search_works e per qualunque altro chiamante generico di +# _request_with_retry. Motivazione: una query generica con molti risultati +# puo' generare decine di batch da risolvere (es. 60 batch per 30 paper con +# referenced_works al cap di 100 ciascuno); con max_retries=3 il caso peggiore +# per singolo batch (assumendo che il read_timeout scatti regolarmente) resta +# nell'ordine delle decine di secondi per tentativo, che su 60 batch si somma +# rapidamente. Con max_retries=1 (2 tentativi totali) il caso peggiore per +# singolo batch si dimezza, riducendo proporzionalmente anche il tetto +# complessivo - in combinazione con il timeout di fase in +# _resolve_references_for_works (vedi etl_pipeline.py), non con l'obiettivo +# di eliminarlo da solo. +RESOLVE_MAX_RETRIES = 1 + +# Budget di default (secondi) per l'INTERA paginazione di search_works, non +# per singola richiesta di pagina. Stesso ruolo di resolve_timeout_seconds in +# get_works_by_ids/_resolve_references_for_works: un secondo argine oltre al +# timeout a tupla di _request_with_retry, indipendente da come si comporta +# la libreria requests in casi limite (connessione tenuta viva artificialmente, +# server lento ma "vivo"). +DEFAULT_FETCH_TIMEOUT_SECONDS = 30 + +# TERZO BUG REALE DIAGNOSTICATO OGGI (debugging log, stesso stile degli altri +# due): _request_with_retry rispettava l'header Retry-After di una risposta +# 429 senza alcun tetto massimo, facendo `time.sleep(float(retry_after))`. +# Diagnosticato concretamente: dopo aver esaurito il budget giornaliero delle +# API OpenAlex durante i test di questa sessione, una richiesta ha ricevuto +# 429 con `Retry-After: 28979` (quasi 8 ore) e il processo e' rimasto +# "bloccato" per ore in un time.sleep() legittimo ma inutilizzabile in un +# contesto interattivo (Shiny). La firma era identica a un socket appeso +# (CPU 0%, stato sleeping, connessione TCP ESTABLISHED del pool keep-alive di +# requests ancora aperta) e per questo era stata scambiata inizialmente per +# un problema di timeout HTTP - non lo era: la risposta 429 arrivava in +# meno di 0.1s, il problema era tutto nello sleep successivo, non tollerato +# ne' dal timeout a tupla di _request_with_retry (si applica solo mentre si +# attende la risposta, non dopo averla ricevuta) ne' dal budget di fase di +# search_works/get_works_by_ids (controllato solo PRIMA di iniziare una nuova +# richiesta, non durante lo sleep interno a un tentativo gia' in corso). +# Nessun retry_after piu' lungo di questo tetto viene piu' onorato per intero: +# se la richiesta fallisce ancora dopo l'attesa limitata e i tentativi +# rimasti, risale normalmente come OpenAlexRequestError (comportamento gia' +# esistente, invariato), che il resto della pipeline e l'handler della pagina +# API sanno gia' gestire mostrando un errore invece di bloccarsi in silenzio. +MAX_RETRY_AFTER_WAIT_SECONDS = 20 + + +class OpenAlexRequestError(Exception): + """Sollevata quando una richiesta a OpenAlex fallisce in modo non recuperabile + (status 4xx diverso da 429, oppure 5xx/timeout dopo l'esaurimento dei retry).""" + pass + + +def search_works(query, max_results=None, filters=None, mailto=None, per_page=MAX_PER_PAGE, + fetch_timeout_seconds=DEFAULT_FETCH_TIMEOUT_SECONDS, start_time=None): + """ + Esegue una ricerca testuale completa su /works, paginando automaticamente + con cursore finche' OpenAlex non restituisce `meta.next_cursor == None` + oppure finche' non e' stato raccolto `max_results` risultati. + + Ogni singola richiesta di pagina passa da _request_with_retry (retry con + backoff esponenziale su 429/5xx/errori di rete, vedi quella funzione). + + LIMITE DI SICUREZZA (debugging log): se `fetch_timeout_seconds` non e' + None, PRIMA di richiedere ogni nuova pagina si controlla il tempo + trascorso da `start_time` (o dall'inizio di questa chiamata, se + start_time non e' fornito): se il budget e' superato, la paginazione si + interrompe con un break (non un'eccezione) e vengono restituiti i + risultati gia' raccolti fino a quel momento (parziali ma utilizzabili), + invece di continuare a chiedere altre pagine indefinitamente. E' un + secondo argine indipendente dal timeout a tupla di _request_with_retry: + diagnosticato un caso reale in cui una singola richiesta HTTP e' rimasta + bloccata piu' di due minuti senza mai sollevare un'eccezione di timeout + (connessione tenuta viva artificialmente) - questo budget limita il danno + anche se il timeout della singola richiesta non dovesse bastare. + + Args: + query: stringa di ricerca libera, mappata sul parametro `search` di OpenAlex. + max_results: numero massimo di risultati da raccogliere complessivamente; + None per scaricare tutti i risultati della query (attenzione: puo' + comportare molte richieste HTTP su query con molti match). + filters: dict opzionale di filtri aggiuntivi (es. {"publication_year": "2020-2023"}), + serializzato nel parametro `filter` di OpenAlex come coppie + "chiave:valore" separate da virgola (AND tra chiavi diverse; per OR + sullo stesso campo il valore va gia' passato nel formato + "a|b|c" da chi costruisce il dict). + mailto: email da passare come parametro `mailto` per rientrare nella + "polite pool" di OpenAlex (rate limit piu' alto e prioritario). + per_page: risultati per pagina richiesti ad OpenAlex ad ogni chiamata + (di default MAX_PER_PAGE=200, il massimo consentito, per minimizzare + il numero di richieste). Esposto come parametro soprattutto per + poterlo abbassare nei test, cosi' da forzare piu' pagine anche con + max_results piccoli. + fetch_timeout_seconds: budget di tempo (secondi) per l'INTERA + paginazione, non per singola richiesta. Default + DEFAULT_FETCH_TIMEOUT_SECONDS (30s). None per nessun limite + (comportamento pre-esistente). + start_time: istante di riferimento (da time.monotonic()) da cui + calcolare il tempo trascorso; se None, si usa l'istante di + ingresso in questa funzione. Permette al chiamante (tipicamente + etl_pipeline.py::_fetch_raw_works) di far partire il cronometro + prima ancora di chiamare search_works. + + Returns: + list[dict]: work grezzi raccolti su tutte le pagine necessarie (o + raccolte prima dell'esaurimento del budget di tempo), nell'ordine + restituito da OpenAlex, troncati a max_results se specificato. + + Raises: + OpenAlexRequestError: se una richiesta di pagina fallisce in modo non + recuperabile (propagata da _request_with_retry). + """ + if max_results is not None and max_results <= 0: + return [] + + if start_time is None: + start_time = time.monotonic() + + effective_per_page = min(per_page, MAX_PER_PAGE) + + results = [] + cursor = "*" + + while cursor is not None: + if fetch_timeout_seconds is not None and (time.monotonic() - start_time) > fetch_timeout_seconds: + logger.warning( + "search_works: budget di %.1fs esaurito, interrotta la paginazione dopo %d risultati raccolti", + fetch_timeout_seconds, len(results), + ) + break + + params = { + "search": query, + "per-page": effective_per_page, + "cursor": cursor, + } + if mailto: + params["mailto"] = mailto + if filters: + params["filter"] = _serialize_filters(filters) + + response = _request_with_retry(WORKS_ENDPOINT, params) + payload = response.json() + + page_results = payload.get("results", []) + if not page_results: + break + + results.extend(page_results) + + if max_results is not None and len(results) >= max_results: + results = results[:max_results] + break + + cursor = (payload.get("meta") or {}).get("next_cursor") + + return results + + +def _serialize_filters(filters): + """ + Funzione interna: serializza un dict di filtri nel formato query-string + atteso dal parametro `filter` di OpenAlex: coppie "chiave:valore" separate + da virgola (semantica AND tra chiavi diverse). La sintassi OR su uno stesso + campo (`chiave:valore1|valore2`) va gia' incapsulata nel valore passato per + quella chiave da chi costruisce il dict `filters`. + + Args: + filters: dict[str, str] di filtri OpenAlex. + + Returns: + str: valore da assegnare al parametro `filter` nella query string. + """ + return ",".join(f"{key}:{value}" for key, value in filters.items()) + + +def get_work_by_id(openalex_id, mailto=None): + """ + Recupera un singolo "work" OpenAlex dato il suo ID (short form "W123..." o URL + completo "https://openalex.org/W123..."). + + Args: + openalex_id: identificatore OpenAlex del work da recuperare. + mailto: email per la polite pool. + + Returns: + dict | None: l'oggetto "work" grezzo, oppure None se non trovato (404). + + Raises: + OpenAlexRequestError: per errori diversi da 404. + """ + raise NotImplementedError + + +def get_works_by_ids(ids, mailto=None, batch_size=DEFAULT_BATCH_SIZE, resolve_timeout_seconds=None, start_time=None): + """ + Risolve in batch una lista di ID OpenAlex verso i rispettivi oggetti "work" + completi, usando il filtro OR `openalex_id:ID1|ID2|...` supportato da /works + (fino a 50 ID per chiamata, limite imposto dall'API OpenAlex) per minimizzare + il numero di richieste HTTP. + + Usato tipicamente da openalex_mapper.py::format_cr_column per risolvere il + contenuto di `referenced_works` (che in OpenAlex e' solo una lista di ID) + nelle citazioni leggibili richieste dalla colonna CR dello schema WoS-style. + + Un batch che fallisce in modo persistente (dopo tutti i retry di + _request_with_retry) NON interrompe la risoluzione degli altri batch: + l'errore viene loggato e gli ID di quel batch restano semplicemente assenti + dal dizionario restituito, cosi' come gli ID non trovati da OpenAlex. Il + chiamante non puo' distinguere "non trovato" da "batch fallito" guardando + solo il dizionario risultato: se questa distinzione servisse a valle, andra' + aggiunta separatamente (es. restituendo anche la lista di ID falliti). + + Ogni chiamata HTTP verso un batch usa RESOLVE_MAX_RETRIES (1 ri-tentativo, + non i 3 di default) invece del default di _request_with_retry: qui i batch + possono essere decine per una singola risoluzione (vedi + etl_pipeline.py::_resolve_references_for_works), quindi il costo peggiore + per singolo batch va tenuto basso deliberatamente, a differenza di + search_works che chiama _request_with_retry con i retry di default. + + Se resolve_timeout_seconds e' specificato, PRIMA di iniziare ogni nuovo + batch si controlla il tempo trascorso da start_time (o dall'inizio di + questa chiamata, se start_time non e' fornito): se il budget e' superato, + il loop si interrompe con un break (non un'eccezione) e viene restituito + il dizionario parziale gia' risolto fino a quel momento. Gli ID dei batch + non ancora processati restano semplicemente assenti dal risultato, con lo + stesso effetto pratico di un batch fallito: format_cr_column ricadra' sul + fallback a ID nudo per quei riferimenti. + + Args: + ids: lista di ID OpenAlex (forma short "W123..." o URL completo + "https://openalex.org/W123...") da risolvere. Duplicati e ID + vuoti/None vengono ignorati. + mailto: email per la polite pool. + batch_size: numero massimo di ID per chiamata; la lista (deduplicata e + normalizzata) viene spezzata in chunk di questa dimensione. + resolve_timeout_seconds: budget di tempo (secondi) per l'INTERA + risoluzione, non per singolo batch. None (default) significa + nessun limite: tutti i batch vengono processati indipendentemente + dal tempo impiegato. + start_time: istante di riferimento (da time.monotonic()) da cui + calcolare il tempo trascorso; se None, si usa l'istante di ingresso + in questa funzione. Permette al chiamante (tipicamente + etl_pipeline.py::_resolve_references_for_works) di far partire il + cronometro prima ancora di chiamare get_works_by_ids. + + Returns: + dict[str, dict]: mappa da ID OpenAlex normalizzato (short form) al + relativo oggetto "work" grezzo. Gli ID non risolvibili (non trovati da + OpenAlex, appartenenti a un batch fallito dopo i retry, oppure mai + raggiunti per esaurimento del budget di tempo) sono semplicemente + assenti dal risultato. + """ + normalized_ids = [] + seen = set() + for raw_id in ids or []: + normalized = _normalize_openalex_id(raw_id) + if normalized and normalized not in seen: + seen.add(normalized) + normalized_ids.append(normalized) + + if start_time is None: + start_time = time.monotonic() + + resolved = {} + for batch_start in range(0, len(normalized_ids), batch_size): + if resolve_timeout_seconds is not None and (time.monotonic() - start_time) > resolve_timeout_seconds: + logger.warning( + "get_works_by_ids: budget di %.1fs esaurito, interrotto dopo %d/%d ID risolti " + "(%d batch rimanenti non processati)", + resolve_timeout_seconds, len(resolved), len(normalized_ids), + (len(normalized_ids) - batch_start + batch_size - 1) // batch_size, + ) + break + + batch = normalized_ids[batch_start:batch_start + batch_size] + params = { + "filter": "openalex_id:" + "|".join(batch), + # per-page deve coprire l'intero batch, altrimenti si rischia di + # ricevere solo i primi DEFAULT_PER_PAGE risultati del filtro OR. + "per-page": len(batch), + } + if mailto: + params["mailto"] = mailto + + try: + response = _request_with_retry(WORKS_ENDPOINT, params, max_retries=RESOLVE_MAX_RETRIES) + except OpenAlexRequestError as exc: + logger.error( + "get_works_by_ids: batch di %d ID fallito dopo i retry (primi ID: %s): %s", + len(batch), batch[:3], exc, + ) + continue + + payload = response.json() + for work in payload.get("results", []): + work_id = _normalize_openalex_id(work.get("id")) + if work_id: + resolved[work_id] = work + + return resolved + + +def _normalize_openalex_id(openalex_id_or_url): + """ + Funzione interna: normalizza un ID OpenAlex alla forma short (es. "W123456789"), + accettando sia la forma short che l'URL completo ("https://openalex.org/W123456789"). + + Args: + openalex_id_or_url: ID OpenAlex in una qualsiasi delle due forme, oppure + None/stringa vuota. + + Returns: + str: ID in forma short, normalizzato; stringa vuota se + openalex_id_or_url e' None/vuoto. + """ + if not openalex_id_or_url: + return "" + return openalex_id_or_url.rsplit("/", 1)[-1] + + +def _request_with_retry(url, params, max_retries=DEFAULT_MAX_RETRIES, backoff_factor=DEFAULT_BACKOFF_FACTOR, timeout=DEFAULT_TIMEOUT): + """ + Funzione interna: esegue una GET HTTP con retry ed exponential backoff. + + Riprova la richiesta in caso di: + - errori di rete/timeout, + - HTTP 429 (rate limit), rispettando l'header Retry-After se presente ma + con un tetto a MAX_RETRY_AFTER_WAIT_SECONDS (un server puo' chiedere + un'attesa di ore, vedi il commento su quella costante; altrimenti + backoff esponenziale), + - HTTP 5xx (errori transitori lato server). + + Non riprova su errori 4xx diversi da 429 (es. 400/404), che vengono considerati + definitivi e propagati immediatamente come OpenAlexRequestError. + + Args: + url: URL completo della richiesta. + params: dict di query string da passare a requests.get. + max_retries: numero massimo di RI-tentativi dopo il primo (quindi al + massimo max_retries + 1 richieste HTTP totali). + backoff_factor: fattore moltiplicativo per il tempo di attesa tra un + tentativo e il successivo (attesa = backoff_factor ** tentativo). + timeout: tupla (connect_timeout, read_timeout) in secondi, passata + direttamente a requests.get. NON e' un tetto sul tempo totale + della richiesta: read_timeout e' il silenzio massimo tollerato + tra un chunk di risposta e il successivo (vedi DEFAULT_TIMEOUT + per il perche' di questa distinzione). + + Returns: + requests.Response: la risposta HTTP con status < 400. + + Raises: + OpenAlexRequestError: su errore 4xx diverso da 429 (nessun retry), oppure + dopo l'esaurimento dei retry per 429/5xx/errori di rete, con status + code e corpo della risposta inclusi nel messaggio per facilitare il debug. + """ + last_error = None + + for attempt in range(max_retries + 1): + try: + response = requests.get(url, params=params, timeout=timeout) + except (requests.exceptions.Timeout, requests.exceptions.ConnectionError) as exc: + last_error = exc + if attempt == max_retries: + raise OpenAlexRequestError( + f"Richiesta a {url} fallita dopo {max_retries + 1} tentativi: {exc}" + ) from exc + time.sleep(backoff_factor ** attempt) + continue + + if response.status_code < 400: + return response + + if response.status_code == 429 or response.status_code >= 500: + last_error = OpenAlexRequestError( + f"HTTP {response.status_code} da {url}: {response.text[:500]}" + ) + if attempt == max_retries: + raise last_error + + retry_after = response.headers.get("Retry-After") + wait_seconds = backoff_factor ** attempt + if retry_after is not None: + try: + # Tetto a MAX_RETRY_AFTER_WAIT_SECONDS: un server puo' + # legittimamente chiedere di attendere ore (visto in + # produzione con un 429 da budget esaurito e + # Retry-After: 28979), ma un'attesa cosi' lunga non e' + # utilizzabile in un contesto interattivo - vedi il + # commento su MAX_RETRY_AFTER_WAIT_SECONDS per il dettaglio. + wait_seconds = min(float(retry_after), MAX_RETRY_AFTER_WAIT_SECONDS) + except ValueError: + pass + time.sleep(wait_seconds) + continue + + # 4xx diverso da 429: errore considerato definitivo, nessun retry. + raise OpenAlexRequestError( + f"HTTP {response.status_code} da {url}: {response.text[:500]}" + ) + + # Non raggiungibile in condizioni normali (il loop ritorna o solleva ad ogni + # iterazione), presente solo per robustezza. + raise OpenAlexRequestError(f"Richiesta a {url} fallita: {last_error}") diff --git a/www/services/openalex_mapper.py b/www/services/openalex_mapper.py new file mode 100644 index 000000000..a1d66bbe2 --- /dev/null +++ b/www/services/openalex_mapper.py @@ -0,0 +1,1166 @@ +""" +Mapping OpenAlex -> schema a 34 colonne WoS-style. + +Speculare a www/services/format_functions.py (che copre WoS/Scopus/Dimensions/ +The_Lens/PubMed/Cochrane): qui e' definita una sola sorgente, "OpenAlex", con una +funzione format_XX_column per ciascuna delle 34 colonne dello schema bibliometrix, +piu' una funzione di orchestrazione map_work_to_record che le combina in un unico +record cosi' come fa il blocco `entry_data = {...}` in format_functions.py::process_single_file. + +Differenze principali rispetto alla pipeline WoS storica, da tenere presenti in +fase di implementazione (vedi analisi fatta in conversazione): +- authorships[].institutions[] e' gia' strutturato (niente split posizionali su + stringhe di affiliazione ne' whitelist di paesi come in metatagextraction.py::AU_CO). +- abstract_inverted_index va ricostruito in una stringa lineare prima di popolare AB. +- referenced_works e' solo una lista di ID: per popolare CR con citazioni leggibili + serve l'output di openalex_client.get_works_by_ids (parametro resolved_references). +- cited_by_count e' gia' un int (a differenza di TC in format_tc_column, che restava + stringa quando proveniva da WoS .txt/.ciw). +- alcune colonne dello schema WoS-style non hanno, per decisione esplicita presa in + fase di design, un equivalente popolato in questa pipeline: EM, FU, FX, JI, OI, + PU, RP, SC restituiscono sempre stringa vuota (vedi i rispettivi format_XX_column + per la motivazione puntuale), anche quando OpenAlex esporrebbe un dato parziale + utilizzabile (es. RP da `is_corresponding`, PU da `host_organization_name`). JI + vuota e' pero' intenzionale anche nella pipeline WoS storica per le sorgenti + senza abbreviazione (vedi format_ji_column): non e' un dato mancante, e' + il segnale che fa scattare il fallback su SO dentro metatagextraction.py::SR. +- SN (ISSN) deriva da `primary_location.source.issn_l`, spostato qui da JI dopo + aver verificato che SN e' il campo ISSN in tutte le sorgenti storiche + (format_functions.py::format_sn_column). +- SR (Short Reference) NON ha una format_sr_column(work) a livello di singolo + record, a differenza delle altre colonne: la funzione esistente riusata per + calcolarla, metatagextraction.py::SR(M), richiede l'INTERA collezione gia' + in forma di DataFrame per poter deduplicare correttamente i valori ripetuti + (suffissi "-a", "-b", ...). Vedi compute_sr_for_records piΓΉ in basso e + etl_pipeline.py::_compute_calculated_fields per come viene usata. +""" + +from .utils import * +from .metatagextraction import SR + + +# Numero massimo di referenced_works considerati da format_cr_column per un +# singolo work. Decisione presa in questo punto dello sviluppo (non +# preesistente): alcuni work OpenAlex citano centinaia di referenze, e +# risolverle/formattarle tutte avrebbe un costo (chiamate batch a +# get_works_by_ids, dimensione di CR) sproporzionato rispetto al beneficio per +# un campo che nello schema WoS-style e' comunque una lista informativa, non +# esaustiva per definizione in molte fonti. +MAX_REFERENCED_WORKS = 100 + +# Mappa dichiarativa colonna WoS-style -> percorso/estrattore OpenAlex. +# Pensata come documentazione leggibile del mapping (colonna -> dove si trova il +# dato in un oggetto "work" OpenAlex), non necessariamente usata a runtime. +# Popolata in fase di implementazione con il mapping discusso in conversazione, +# es.: {"TI": "title", "PY": "publication_year", "SO": "primary_location.source.display_name", ...} +OPENALEX_FIELD_MAP: dict = {} + +# Mappa OpenAlex `type` (vocabolario controllato, vedi +# https://docs.openalex.org/api-entities/works/work-object#type) -> vocabolario +# WoS-style per la colonna DT. Copre i type piu' comuni; un type OpenAlex non +# presente qui non e' un errore: format_dt_column ricade sul valore originale +# capitalizzato invece di perdere l'informazione (vedi format_dt_column). +OPENALEX_TYPE_TO_WOS_DT: dict = { + "article": "Article", + "review": "Review", + "book-chapter": "Book Chapter", + "preprint": "Preprint", + "dataset": "Dataset", + "dissertation": "Thesis", + "book": "Book", + "editorial": "Editorial Material", + "letter": "Letter", + "erratum": "Correction", + "report": "Report", + "peer-review": "Peer Review", + "paratext": "Paratext", + "standard": "Standard", + "grant": "Grant", + "supplementary-materials": "Supplementary Materials", + "reference-entry": "Reference Entry", + "other": "Other", +} + +# Lookup ISO 3166-1 alpha-2 -> nome paese, nella STESSA convenzione di naming usata +# dalla whitelist in www/static/countries.txt (consumata da +# metatagextraction.py::AU_CO/AU1_CO/AU_UN). I nomi qui DEVONO combaciare +# esattamente con le righe di quel file, cosi' che una stringa C1 costruita da +# format_c1_column resti riconoscibile dal matching a valle +# (`c1.split(",")[-1].strip().upper()` + ricerca `\b\b` nella whitelist) +# senza dover toccare metatagextraction.py. +# +# Note sulle scelte fatte per allinearsi a countries.txt: +# - "US" -> "UNITED STATES" (non "USA": entrambe le forme sono in whitelist, ma +# AU_CO() normalizza comunque "UNITED STATES" -> "USA" a valle, quindi il +# risultato finale coincide). +# - "GB" -> "UNITED KINGDOM" (in whitelist esistono anche ENGLAND/SCOTLAND/WALES/ +# NORTH IRELAND come alias storici WoS, ma non sono codici ISO e OpenAlex non +# li restituisce mai come country_code). +# - "RU" -> "RUSSIA" (non "RUSSIAN FEDERATION", assente dalla whitelist). +# - "MK" -> "NORTH MACEDONIA" (nome ISO corrente; "MACEDONIA" resta in whitelist +# come alias storico ma non e' il target di questa lookup). +# - "TW" -> "TAIWAN" (AU_CO() normalizza comunque "TAIWAN" -> "CHINA" a valle). +# - "CD" -> "CONGO" (la whitelist ha una sola voce "CONGO", senza distinzione +# Congo-Brazzaville/Congo-Kinshasa; stessa approssimazione gia' presente li'). +ISO_COUNTRY_CODE_TO_NAME: dict = { + "AF": "AFGHANISTAN", "AL": "ALBANIA", "DZ": "ALGERIA", "AD": "ANDORRA", + "AO": "ANGOLA", "AG": "ANTIGUA", "AR": "ARGENTINA", "AM": "ARMENIA", + "AU": "AUSTRALIA", "AT": "AUSTRIA", "AZ": "AZERBAIJAN", "BS": "BAHAMAS", + "BH": "BAHRAIN", "BD": "BANGLADESH", "BB": "BARBADOS", "BY": "BELARUS", + "BE": "BELGIUM", "BZ": "BELIZE", "BJ": "BENIN", "BT": "BHUTAN", + "BO": "BOLIVIA", "BA": "BOSNIA", "BW": "BOTSWANA", "BR": "BRAZIL", + "BN": "BRUNEI", "BG": "BULGARIA", "BF": "BURKINA FASO", "BI": "BURUNDI", + "CV": "CABO VERDE", "KH": "CAMBODIA", "CM": "CAMEROON", "CA": "CANADA", + "CF": "CENTRAL AFRICAN REPUBLIC", "TD": "CHAD", "CL": "CHILE", "CN": "CHINA", + "CO": "COLOMBIA", "KM": "COMOROS", "CG": "CONGO", "CD": "CONGO", + "CR": "COSTA RICA", "CI": "COTE D'IVOIRE", "HR": "CROATIA", "CU": "CUBA", + "CY": "CYPRUS", "CZ": "CZECH REPUBLIC", "DK": "DENMARK", "DJ": "DJIBOUTI", + "DM": "DOMINICA", "DO": "DOMINICAN REPUBLIC", "EC": "ECUADOR", "EG": "EGYPT", + "SV": "EL SALVADOR", "GQ": "EQUATORIAL GUINEA", "ER": "ERITREA", + "EE": "ESTONIA", "ET": "ETHIOPIA", "FO": "FAROE", "FJ": "FIJI", + "FI": "FINLAND", "FR": "FRANCE", "GA": "GABON", "GM": "GAMBIA", + "GE": "GEORGIA", "DE": "GERMANY", "GH": "GHANA", "GR": "GREECE", + "GU": "GUAM", "GT": "GUATEMALA", "GN": "GUINEA", "GW": "GUINEA-BISSAU", + "HT": "HAITI", "HN": "HONDURAS", "HK": "HONG KONG", "HU": "HUNGARY", + "IS": "ICELAND", "IN": "INDIA", "ID": "INDONESIA", "IR": "IRAN", + "IQ": "IRAQ", "IE": "IRELAND", "IL": "ISRAEL", "IT": "ITALY", + "JM": "JAMAICA", "JP": "JAPAN", "JO": "JORDAN", "KZ": "KAZAKHSTAN", + "KE": "KENYA", "KI": "KIRIBATI", "KR": "KOREA", "XK": "KOSOVO", + "KW": "KUWAIT", "KG": "KYRGYZSTAN", "LA": "LAOS", "LV": "LATVIA", + "LB": "LEBANON", "LS": "LESOTHO", "LR": "LIBERIA", "LY": "LIBYA", + "LI": "LIECHTENSTEIN", "LT": "LITHUANIA", "LU": "LUXEMBOURG", + "MK": "NORTH MACEDONIA", "MG": "MADAGASCAR", "MW": "MALAWI", + "MY": "MALAYSIA", "MV": "MALDIVES", "ML": "MALI", "MT": "MALTA", + "MH": "MARSHALL ISLANDS", "MR": "MAURITANIA", "MU": "MAURITIUS", + "MX": "MEXICO", "FM": "MICRONESIA", "MD": "MOLDOVA", "MC": "MONACO", + "MN": "MONGOLIA", "ME": "MONTENEGRO", "MA": "MOROCCO", "MZ": "MOZAMBIQUE", + "MM": "MYANMAR", "NA": "NAMIBIA", "NR": "NAURU", "NP": "NEPAL", + "NL": "NETHERLANDS", "NZ": "NEW ZEALAND", "NI": "NICARAGUA", "NE": "NIGER", + "NG": "NIGERIA", "KP": "NORTH KOREA", "NO": "NORWAY", "OM": "OMAN", + "PK": "PAKISTAN", "PW": "PALAU", "PA": "PANAMA", "PG": "PAPUA NEW GUINEA", + "PY": "PARAGUAY", "PE": "PERU", "PH": "PHILIPPINES", "PL": "POLAND", + "PT": "PORTUGAL", "QA": "QATAR", "RO": "ROMANIA", "RU": "RUSSIA", + "RW": "RWANDA", "KN": "SAINT KITTS AND NEVIS", "LC": "SAINT LUCIA", + "WS": "SAMOA", "SM": "SAN MARINO", "ST": "SAO TOME AND PRINCIPE", + "SA": "SAUDI ARABIA", "SN": "SENEGAL", "RS": "SERBIA", "SC": "SEYCHELLES", + "SL": "SIERRA LEONE", "SG": "SINGAPORE", "SK": "SLOVAKIA", "SI": "SLOVENIA", + "SB": "SOLOMON ISLANDS", "SO": "SOMALIA", "ZA": "SOUTH AFRICA", + "SS": "SOUTH SUDAN", "ES": "SPAIN", "LK": "SRI LANKA", "SD": "SUDAN", + "SR": "SURINAME", "SZ": "SWAZILAND", "SE": "SWEDEN", "CH": "SWITZERLAND", + "SY": "SYRIA", "TW": "TAIWAN", "TJ": "TAJIKISTAN", "TZ": "TANZANIA", + "TH": "THAILAND", "TG": "TOGO", "TO": "TONGA", "TT": "TRINIDAD AND TOBAGO", + "TN": "TUNISIA", "TR": "TURKEY", "TM": "TURKMENISTAN", "UG": "UGANDA", + "UA": "UKRAINE", "AE": "UNITED ARAB EMIRATES", "GB": "UNITED KINGDOM", + "US": "UNITED STATES", "UY": "URUGUAY", "UZ": "UZBEKISTAN", "VU": "VANUATU", + "VA": "VATICANO", "VE": "VENEZUELA", "VN": "VIETNAM", "YE": "YEMEN", + "ZM": "ZAMBIA", "ZW": "ZIMBABWE", +} + + +def map_work_to_record(work, resolved_references=None): + """ + Funzione di orchestrazione: converte un singolo oggetto "work" OpenAlex grezzo + in un record (dict) nello schema a 34 colonne WoS-style, chiamando in sequenza + tutte le format_XX_column definite in questo modulo. + + Analoga al blocco `entry_data = {...}` in + www/services/format_functions.py::process_single_file, ma con un'unica sorgente + (OpenAlex) invece del branching multi-sorgente/multi-formato usato li'. + + Non calcola i campi che richiedono l'intera collezione (es. SR, che necessita + di deduplicare "Autore, Anno, Rivista" su tutto il dataset): quello e' compito + di etl_pipeline.py::_compute_calculated_fields, eseguito dopo questa funzione. + + Args: + work: dict, singolo oggetto "work" grezzo restituito da OpenAlex + (vedi openalex_client.search_works). + resolved_references: dict[str, dict] opzionale, mappa ID OpenAlex -> work + risolto, usata da format_cr_column per costruire citazioni leggibili a + partire da `referenced_works`. None se la risoluzione e' stata saltata + (in quel caso CR conterra' presumibilmente i soli ID grezzi). + + Returns: + dict: record con esattamente le chiavi di `columns` (lista canonica + definita in www/services/utils.py) meno "SR" β€” 33 chiavi in totale. + + Raises: + ValueError: se il dict costruito internamente non coincide esattamente + (ne' per difetto ne' per eccesso) con `set(columns) - {"SR"}| β€” + indica una regressione tra questa funzione e la lista canonica in + utils.py, non un problema sui dati del work in input. + """ + record = { + "AB": format_ab_column(work), + "AF": format_af_column(work), + "AU": format_au_column(work), + "AU_UN": format_au_un_column(work), + "AU1_UN": format_au1_un_column(work), + "BP": format_bp_column(work), + "EP": format_ep_column(work), + "CR": format_cr_column(work, resolved_references), + "C1": format_c1_column(work), + "DB": format_db_column(work), + "DE": format_de_column(work), + "DI": format_di_column(work), + "DT": format_dt_column(work), + "EM": format_em_column(work), + "FU": format_fu_column(work), + "FX": format_fx_column(work), + "IS": format_is_column(work), + "JI": format_ji_column(work), + "ID": format_id_column(work), + "LA": format_la_column(work), + "OA": format_oa_column(work), + "OI": format_oi_column(work), + "PMID": format_pmid_column(work), + "PU": format_pu_column(work), + "PY": format_py_column(work), + "RP": format_rp_column(work), + "SC": format_sc_column(work), + "SN": format_sn_column(work), + "SO": format_so_column(work), + "TC": format_tc_column(work), + "TI": format_ti_column(work), + "UT": format_ut_column(work), + "VL": format_vl_column(work), + } + + expected_keys = set(columns) - {"SR"} + actual_keys = set(record.keys()) + if actual_keys != expected_keys: + missing = sorted(expected_keys - actual_keys) + extra = sorted(actual_keys - expected_keys) + raise ValueError( + "map_work_to_record: il record prodotto non coincide con lo schema " + "canonico (www/services/utils.py::columns) meno SR. " + f"Mancanti: {missing}. Extra: {extra}." + ) + + return record + + +def format_ab_column(work): + """ + Colonna AB (Abstract). Ricostruisce il testo lineare dell'abstract a partire da + `work["abstract_inverted_index"]` (dict parola -> lista di posizioni), che in + OpenAlex sostituisce l'abstract come stringa unica presente in WoS. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: abstract ricostruito, oppure stringa vuota se `abstract_inverted_index` + e' assente/None (es. per motivi di copyright, come spesso accade in OpenAlex). + """ + return _reconstruct_abstract(work.get("abstract_inverted_index")) + + +def format_af_column(work): + """ + Colonna AF (Authors Full Name). Per la sorgente OpenAlex coincide + esattamente con AU (vedi format_au_column e la nota nel suo docstring): + entrambe derivano da `work["authorships"][].author.display_name` senza + alcuna trasformazione. Questa funzione delega direttamente a + format_au_column invece di duplicarne la logica. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + list[str]: identico all'output di format_au_column(work). + """ + return format_au_column(work) + + +def format_au_column(work): + """ + Colonna AU (Author/s). A differenza di format_au_column in + format_functions.py (che normalizza WoS/Scopus/ecc. nel formato "COGNOME + Iniziali"), qui si usa direttamente `work["authorships"][].author.display_name` + cosi' come restituito da OpenAlex, senza alcuna logica di split cognome/nome: + scelta deliberata per evitare euristiche fragili sui nomi (es. nomi composti, + prefissi, ordini cognome-nome non occidentali) quando il dato "display_name" + e' gia' disponibile in forma leggibile. + + Nota: con questa scelta AU e AF risultano uguali per la sorgente OpenAlex + (entrambi il nome completo dell'autore), a differenza della pipeline WoS + storica dove sono formati distinti ("Cognome I." vs "Cognome, Nome completo"). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + list[str]: un elemento per autore, pari a `author.display_name`, + nell'ordine restituito da `work["authorships"]`. Autori senza `author` o + senza `display_name` vengono omessi. + """ + authors = [] + for authorship in work.get("authorships") or []: + author = authorship.get("author") or {} + display_name = author.get("display_name") + if display_name: + authors.append(display_name) + return authors + + +def format_au_un_column(work): + """ + Colonna AU_UN (Authors University/Institution). Estrae le istituzioni da + `work["authorships"][].institutions[].display_name`, gia' strutturate e + disambiguate da OpenAlex (a differenza di AU_UN in metatagextraction.py, che + deve inferire l'istituzione da una stringa di affiliazione grezza tramite una + whitelist di tag come "UNIV", "COLL", ecc.). + + Restituisce list[str] (non una singola stringa ";"-separated), per coerenza + con cio' che si aspettano i consumer a valle sull'output della pipeline di + import diretta (es. functions/get_affiliationproductionovertime.py e + functions/get_relevantaffiliations.py, che trattano AU_UN come lista per + riga tramite `.apply(...)`/`.explode()`). + + Deduplicazione: se piu' autori condividono la stessa istituzione (stesso + `display_name`), questa compare una sola volta nell'output, preservando + l'ordine di prima apparizione. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + list[str]: nomi delle istituzioni distinte associate agli autori del work, + lista vuota se `work["authorships"]` e' assente/vuoto o nessun autore ha + institutions. + """ + seen = set() + universities = [] + for authorship in work.get("authorships") or []: + for institution in authorship.get("institutions") or []: + name = institution.get("display_name") + if name and name not in seen: + seen.add(name) + universities.append(name) + return universities + + +def format_au1_un_column(work): + """ + Colonna AU1_UN (Institution of the First Author). A differenza di AU_UN, + restituisce una singola stringa (non una lista), per coerenza con + format_au1_un_column in format_functions.py (che restituisce sempre una + stringa, es. `str(entry.get('C3','')).split("; ")[0]`) e con l'etichetta + "First Author University" (singolare) usata in functions/get_table.py. + + Il primo autore e' individuato cercando `author_position == "first"` in + `work["authorships"]`; se il campo non e' presente/valorizzato su nessun + elemento, si ricade sul primo elemento della lista `authorships` (che in + pratica coincide quasi sempre con l'autore in posizione "first"). + + Se il primo autore ha piu' institutions, viene presa solo la prima (stessa + scelta "solo il primo valore" fatta da format_au1_un_column in + format_functions.py per WoS, che tiene solo l'indice 0 dopo lo split su C3). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: nome dell'istituzione del primo autore, "" se non disponibile + (nessun autore, primo autore senza institutions, o institution senza + display_name). + """ + authorships = work.get("authorships") or [] + if not authorships: + return "" + + first_authorship = next( + (a for a in authorships if a.get("author_position") == "first"), + authorships[0], + ) + + institutions = first_authorship.get("institutions") or [] + if not institutions: + return "" + + return institutions[0].get("display_name") or "" + + +def format_bp_column(work): + """ + Colonna BP (Beginning Page). Deriva da `work["biblio"]["first_page"]`, con + accesso robusto ai campi annidati: se `work["biblio"]` e' assente/None, + viene trattato come dict vuoto invece di sollevare AttributeError/KeyError. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: numero di pagina iniziale, "" se assente (comune per preprint, + come visto nell'esempio arXiv analizzato in conversazione). + """ + biblio = work.get("biblio") or {} + return biblio.get("first_page") or "" + + +def format_ep_column(work): + """ + Colonna EP (Ending Page). Deriva da `work["biblio"]["last_page"]`, con + accesso robusto ai campi annidati: se `work["biblio"]` e' assente/None, + viene trattato come dict vuoto invece di sollevare AttributeError/KeyError. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: numero di pagina finale, "" se assente. + """ + biblio = work.get("biblio") or {} + return biblio.get("last_page") or "" + + +def format_cr_column(work, resolved_references=None): + """ + Colonna CR (Cited References). Deriva da `work["referenced_works"]`, che in + OpenAlex e' solo una lista di ID (non stringhe bibliografiche complete come + in WoS). + + Funzione di solo mapping: non fa I/O di rete. `resolved_references` deve + essere gia' stato popolato a monte (tipicamente da una singola chiamata + batch a openalex_client.get_works_by_ids su tutti gli ID di + `referenced_works` di uno o piu' work) e viene passato qui gia' pronto, + per mantenere la separazione tra mapping puro (questo modulo) e I/O di + rete (openalex_client.py). + + Per ogni ID in `work["referenced_works"]` (troncato a MAX_REFERENCED_WORKS + elementi): + - se l'ID e' presente in `resolved_references`, la citazione e' costruita + nello stesso stile "Autore, ANNO, RIVISTA" usato da + metatagextraction.py::SR, riusando format_au_column/format_py_column/ + format_so_column sul work risolto (vedi _format_reference_citation); + - se l'ID NON e' risolvibile (risoluzione fallita, batch andato in errore, + o resolved_references non fornito/None), il riferimento non viene + scartato: l'ID OpenAlex nudo viene incluso cosi' com'e' come fallback, + cosi' l'informazione "esiste un riferimento qui" non va persa in + silenzio anche se non e' stato possibile arricchirla. + + Args: + work: oggetto "work" OpenAlex grezzo. + resolved_references: dict[str, dict] opzionale, mappa ID OpenAlex + normalizzato -> work risolto (vedi openalex_client.get_works_by_ids). + None o {} equivalgono a "nessuna risoluzione disponibile": tutti i + riferimenti ricadono sul fallback ID nudo. + + Returns: + list[str]: una voce per riferimento citato (citazione leggibile o ID + nudo), nell'ordine di `work["referenced_works"]`, troncata a + MAX_REFERENCED_WORKS elementi. Lista vuota se `referenced_works` e' + assente/vuoto. + """ + referenced_ids = (work.get("referenced_works") or [])[:MAX_REFERENCED_WORKS] + resolved_references = resolved_references or {} + + citations = [] + for raw_id in referenced_ids: + if not raw_id: + continue + + normalized_id = raw_id.rsplit("/", 1)[-1] + resolved_work = resolved_references.get(normalized_id) + + if resolved_work: + citations.append(_format_reference_citation(resolved_work)) + else: + citations.append(normalized_id) + + return citations + + +def _format_reference_citation(resolved_work): + """ + Funzione interna: costruisce la citazione leggibile "Autore, ANNO, RIVISTA" + per un singolo work OpenAlex risolto, riusando le format_XX_column gia' + definite in questo modulo (format_au_column, format_py_column, + format_so_column) invece di duplicare la logica di estrazione β€” stesso + stile della formula "FirstAuthors, PY, J9" usata da + metatagextraction.py::SR per il riferimento breve di un documento. + + Usata da format_cr_column. + + Args: + resolved_work: dict, oggetto "work" OpenAlex grezzo del riferimento + citato (gia' risolto tramite openalex_client.get_works_by_ids). + + Returns: + str: "Autore, ANNO, RIVISTA". Il primo autore e' "NA" se il work + risolto non ha autori (stessa convenzione di metatagextraction.py::SR); + RIVISTA puo' comparire come stringa vuota se non disponibile sul work + risolto; ANNO e' sempre presente come stringa numerica (0 se il work + risolto non ha `publication_year`, vedi format_py_column). + """ + authors = format_au_column(resolved_work) + first_author = authors[0] if authors else "NA" + year = format_py_column(resolved_work) # int (vedi format_py_column) -> cast esplicito qui sotto + journal = format_so_column(resolved_work) + return f"{first_author}, {str(year)}, {journal}" + + +def format_c1_column(work): + """ + Colonna C1 (Authors Affiliation). Fonte dati: i campi strutturati OpenAlex + (`authorships[].author.display_name`, `authorships[].institutions[].display_name`, + `authorships[].institutions[].country_code`), NON `raw_affiliation_strings` + (testo grezzo non normalizzato). + + Ogni stringa di output e' costruita nel formato WoS-style + "[NomeAutore] Nome Istituzione, PAESE" (paese in maiuscolo), riproducendo la + convenzione con cui format_c1_column in format_functions.py rimuove il + prefisso "[Autori] " dalle righe C1 di WoS β€” qui il prefisso viene invece + generato ex novo a partire da un dato gia' strutturato. + + Il nome del paese e' ottenuto da ISO_COUNTRY_CODE_TO_NAME, che usa + ESATTAMENTE la stessa convenzione di naming della whitelist in + www/static/countries.txt: questo garantisce che + metatagextraction.py::AU_CO/AU1_CO/AU_UN (che deriva paesi/istituzioni da C1 + facendo `c1.split(",")[-1].strip().upper()` e cercando il risultato nella + whitelist) continui a funzionare invariato anche sulle righe C1 generate da + questa funzione, senza bisogno di modifiche a valle. + + Un'istituzione senza `country_code` riconosciuto in ISO_COUNTRY_CODE_TO_NAME + produce comunque una riga C1 valida, ma senza il segmento finale ", PAESE": + in quel caso il matching a valle in AU_CO/AU1_CO semplicemente non trovera' + un paese per quella affiliazione (stesso comportamento di una riga WoS con + paese mancante/non riconosciuto). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + list[str]: una stringa "[NomeAutore] Nome Istituzione, PAESE" per ogni + combinazione (autore, istituzione) presente in `work["authorships"]`. + Autori senza institutions non producono alcuna riga (nessuna affiliazione + da riportare). Lista vuota se `work["authorships"]` e' assente/vuoto. + """ + affiliations = [] + for authorship in work.get("authorships") or []: + author = authorship.get("author") or {} + author_name = author.get("display_name") + if not author_name: + continue + + for institution in authorship.get("institutions") or []: + institution_name = institution.get("display_name") + if not institution_name: + continue + + country_code = institution.get("country_code") + country_name = ISO_COUNTRY_CODE_TO_NAME.get((country_code or "").upper()) + + if country_name: + affiliation = f"[{author_name}] {institution_name}, {country_name}" + else: + affiliation = f"[{author_name}] {institution_name}" + + affiliations.append(affiliation) + + return affiliations + + +def format_db_column(work): + """ + Colonna DB (Database). Costante: tutti i record prodotti da questo mapper + provengono dalla sorgente OpenAlex. + + Args: + work: oggetto "work" OpenAlex grezzo (non usato, presente solo per + uniformita' di firma con le altre format_XX_column). + + Returns: + str: sempre "OPENALEX". + """ + return "OPENALEX" + + +def format_de_column(work): + """ + Colonna DE (Author Keywords). Deriva da `work["keywords"][].display_name` + (i keyword assegnati algoritmicamente da OpenAlex con score di confidenza, + non keyword scelte dall'autore come in WoS: da segnalare come scostamento + semantico rispetto al significato originale di DE). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + list[str]: keyword estratte da `work["keywords"]`, lista vuota se il + campo e' assente/vuoto o gli elementi non hanno `display_name`. + """ + keywords = [] + for keyword in work.get("keywords") or []: + name = keyword.get("display_name") + if name: + keywords.append(name) + return keywords + + +def format_di_column(work): + """ + Colonna DI (DOI). Deriva da `work["doi"]`, normalizzato rimuovendo il prefisso + URL "https://doi.org/" per ottenere il DOI nudo nel formato atteso dallo schema + WoS-style (es. "10.48550/arxiv.1201.0490"). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: DOI nudo, "" se `work["doi"]` e' assente/None. + """ + doi = work.get("doi") + if not doi: + return "" + prefix = "https://doi.org/" + if doi.startswith(prefix): + return doi[len(prefix):] + return doi + + +def format_dt_column(work): + """ + Colonna DT (Document Type). Deriva da `work["type"]` (vocabolario controllato + OpenAlex, es. "article", "preprint", "book-chapter"), tradotto nel vocabolario + WoS-style (es. "Article", "Review") tramite OPENALEX_TYPE_TO_WOS_DT. + + Se `work["type"]` non e' presente in OPENALEX_TYPE_TO_WOS_DT (type nuovo o + non ancora mappato), il fallback e' il valore OpenAlex originale con la + prima lettera maiuscola (`str.capitalize()`), invece di una stringa vuota: + l'informazione grezza resta comunque visibile/utilizzabile a valle anziche' + andare persa silenziosamente. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: valore DT mappato, il valore originale capitalizzato se non + presente in OPENALEX_TYPE_TO_WOS_DT, "" se `work["type"]` e' assente/None. + """ + work_type = work.get("type") + if not work_type: + return "" + return OPENALEX_TYPE_TO_WOS_DT.get(work_type, work_type.capitalize()) + + +def format_em_column(work): + """ + Colonna EM (Email). Campo non disponibile da OpenAlex (nessun indirizzo email + degli autori esposto dall'API): restituisce sempre stringa vuota per decisione + esplicita, invece di propagare KeyError/AttributeError a valle. + + Args: + work: oggetto "work" OpenAlex grezzo (non usato). + + Returns: + str: sempre "". + """ + return "" + + +def format_fu_column(work): + """ + Colonna FU (Funding Details). Campo non disponibile da OpenAlex per questa + pipeline: restituisce sempre stringa vuota per decisione esplicita. + + Args: + work: oggetto "work" OpenAlex grezzo (non usato). + + Returns: + str: sempre "". + """ + return "" + + +def format_fx_column(work): + """ + Colonna FX (Funding Text). Campo non disponibile da OpenAlex per questa + pipeline: restituisce sempre stringa vuota per decisione esplicita. + + Args: + work: oggetto "work" OpenAlex grezzo (non usato). + + Returns: + str: sempre "". + """ + return "" + + +def format_is_column(work): + """ + Colonna IS (Issue). Deriva da `work["biblio"]["issue"]`, con accesso robusto + ai campi annidati: se `work["biblio"]` e' assente/None, viene trattato come + dict vuoto invece di sollevare AttributeError/KeyError. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: numero di fascicolo, "" se assente. + """ + biblio = work.get("biblio") or {} + return biblio.get("issue") or "" + + +def format_ji_column(work): + """ + Colonna JI (Abbreviated Journal Name). OpenAlex non fornisce + un'abbreviazione standardizzata del nome rivista: restituisce sempre "". + + Non e' un dato perso: metatagextraction.py::SR (riusata da + compute_sr_for_records) tratta esplicitamente JI == "" come "nessuna + abbreviazione disponibile" e ricade sul nome completo della rivista (SO) + per costruire la colonna SR β€” esattamente lo stesso pattern gia' usato + dalla pipeline WoS storica per le sorgenti senza abbreviazione (Dimensions, + The_Lens, Cochrane in format_functions.py::format_ji_column). + + Args: + work: oggetto "work" OpenAlex grezzo (non usato). + + Returns: + str: sempre "". + """ + return "" + + +def format_id_column(work): + """ + Colonna ID (Index/Keywords Plus). Deriva da `work["concepts"][].display_name`. + + Nota concettuale importante: a differenza di Keywords Plus in WoS (termini + estratti algoritmicamente dai titoli delle referenze citate da un articolo), + i "concepts" di OpenAlex sono TOPIC assegnati algoritmicamente al work stesso + tramite un classificatore proprietario di OpenAlex, organizzati in una + gerarchia (`level`) con uno score di confidenza. Sono quindi concettualmente + piu' vicini a una tassonomia/categorizzazione automatica del contenuto che + non a delle "keyword aggiuntive" nel senso WoS del termine: vanno trattati + come un'approssimazione, non come un equivalente semantico di ID. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + list[str]: nomi dei concetti estratti da `work["concepts"]`, lista vuota + se il campo e' assente/vuoto o gli elementi non hanno `display_name`. + """ + concepts = [] + for concept in work.get("concepts") or []: + name = concept.get("display_name") + if name: + concepts.append(name) + return concepts + + +def format_la_column(work): + """ + Colonna LA (Language). Pass-through diretto di `work["language"]` (codice + ISO 639-1, es. "en"), a differenza di WoS dove LA e' tipicamente il nome + esteso della lingua (es. "English"): da segnalare come possibile scostamento + di formato se il resto della pipeline (es. filtri o report) si aspetta il + nome esteso. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: codice lingua ISO 639-1, "" se `work["language"]` e' assente/None. + """ + return work.get("language") or "" + + +def format_oa_column(work): + """ + Colonna OA (Open Access). Deriva da `work["open_access"]["oa_status"]` + (vocabolario controllato OpenAlex: "gold", "green", "hybrid", "bronze", + "closed", ...), pass-through diretto. + + NOTA: implementata qui solo perche' necessaria a map_work_to_record per + produrre tutte le 33 chiavi richieste (34 colonne meno SR) β€” senza di essa + map_work_to_record avrebbe sollevato NotImplementedError su OA. A + differenza di EM/FU/FX/OI/PU/RP/SC/SN (tutte deliberatamente "sempre + vuote" per decisione esplicita presa in conversazione), OA NON e' stata + oggetto della stessa decisione esplicita: va rivista se il formato + desiderato e' diverso da un semplice pass-through di `oa_status` + (es. un booleano derivato da `is_oa`, invece della stringa di stato). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: valore di `oa_status`, "" se `open_access`/`oa_status` sono + assenti/None. + """ + open_access = work.get("open_access") or {} + return open_access.get("oa_status") or "" + + +def format_oi_column(work): + """ + Colonna OI (ORCID). Campo non disponibile da OpenAlex per questa pipeline: + restituisce sempre stringa vuota per decisione esplicita. + + Args: + work: oggetto "work" OpenAlex grezzo (non usato). + + Returns: + str: sempre "". + """ + return "" + + +def format_pmid_column(work): + """ + Colonna PMID (PubMed ID). Deriva da `work["ids"]["pmid"]`, se presente (non + tutti i work OpenAlex hanno un ID PubMed associato), normalizzato all'ID nudo + (senza l'eventuale prefisso URL "https://pubmed.ncbi.nlm.nih.gov/"). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: PMID nudo, "" se `work["ids"]["pmid"]` e' assente/None. + """ + ids = work.get("ids") or {} + pmid = ids.get("pmid") + if not pmid: + return "" + return pmid.rsplit("/", 1)[-1] + + +def format_pu_column(work): + """ + Colonna PU (Publisher). Campo non disponibile da OpenAlex per questa + pipeline: restituisce sempre stringa vuota per decisione esplicita (pur + essendo `primary_location.source.host_organization_name` potenzialmente + disponibile, non viene usato in questa versione della pipeline). + + Args: + work: oggetto "work" OpenAlex grezzo (non usato). + + Returns: + str: sempre "". + """ + return "" + + +def format_py_column(work): + """ + Colonna PY (Publication Year). Deriva da `work["publication_year"]` + (gia' un int lato OpenAlex). + + BUG DOCUMENTATO E RISOLTO (rilevante per la relazione finale, sezione + "Weak or inconsistent type enforcement"): questa funzione restituiva in + origine str(publication_year), classificando PY come STRING in + type_contracts.py::COLUMN_SPECS. Il ragionamento iniziale era "replicare + il tipo grezzo della pipeline WoS storica", che infatti restituisce anche + li' una stringa (format_functions.py::format_py_column). Il problema: + nella pipeline storica quella stringa diventa int64 "gratis" perche' + functions/get_data.py costruisce il DataFrame con + `pd.read_json(StringIO(json))`, che applica inferenza automatica di tipo + alle stringhe numeriche; la nostra etl_pipeline.py::_build_dataframe usa + invece `pd.DataFrame(records)` diretto, che non fa questa inferenza β€” con + PY=str, functions/get_annualproduction.py andava in TypeError su + `range(min_year, max_year + 1)` (somma int su stringa). + + Verificato con grep su format_functions.py e su tutte le functions/*.py: + nessun consumer richiede PY come stringa (nessuno slicing, nessun + accessor .str, nessuna concatenazione diretta); numerosi consumer lo + richiedono esplicitamente numerico (min/max, range, confronti aritmetici, + groupby, np.linspace/pd.cut), e 4 di essi fanno gia' un cast difensivo + `pd.to_numeric(df['PY'], errors='coerce')` proprio per la stessa ragione + (get_authorlocalimpact.py, get_authorproductionovertime.py, + get_sourceslocalimpact.py, get_thematicevolution.py). Corretto qui + restituendo un int nativo, e classificando PY come INTEGER in + type_contracts.py::COLUMN_SPECS invece di STRING: il DataFrame prodotto + da questa pipeline ottiene cosi' lo stesso dtype numerico che la pipeline + storica ottiene indirettamente tramite il roundtrip JSON. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + int: anno di pubblicazione, 0 se `work["publication_year"]` e' + assente/None/non convertibile (stesso default "assenza" usato per TC + in format_tc_column). + """ + py = work.get("publication_year") + if py is None: + return 0 + try: + return int(py) + except (TypeError, ValueError): + return 0 + + +def format_rp_column(work): + """ + Colonna RP (Correspondence Address/Reprint Author). Per decisione esplicita, + restituisce sempre stringa vuota: OpenAlex espone un flag `is_corresponding` + per autore, ma non un indirizzo di corrispondenza strutturato/affidabile + equivalente a RP in WoS, quindi si evita di costruire un dato di bassa + qualita' a partire da quel flag. + + Args: + work: oggetto "work" OpenAlex grezzo (non usato). + + Returns: + str: sempre "". + """ + return "" + + +def format_sc_column(work): + """ + Colonna SC (Subject Category / Fields of Research). Campo non disponibile da + OpenAlex per questa pipeline: restituisce sempre stringa vuota per decisione + esplicita (pur essendo `topics[].field.display_name` potenzialmente + disponibile, non viene usato in questa versione della pipeline). + + Args: + work: oggetto "work" OpenAlex grezzo (non usato). + + Returns: + str: sempre "". + """ + return "" + + +def format_sn_column(work): + """ + Colonna SN (ISSN). Deriva da + `work["primary_location"]["source"]["issn_l"]`, con accesso robusto ai + campi annidati: `primary_location` e/o `source` possono essere None per + interi record (es. preprint su repository senza ISSN, come l'esempio + arXiv analizzato in conversazione), nel qual caso vengono trattati come + dict vuoti invece di sollevare AttributeError/KeyError. + + Verificato contro format_functions.py::format_sn_column: SN e' il campo + ISSN in tutte le sorgenti storiche (WoS, PubMed, Scopus, The_Lens) β€” questo + valore era prima erroneamente assegnato a JI (Abbreviated Journal Name), + corretto qui. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: ISSN-L della sorgente, "" se `primary_location`/`source`/`issn_l` + sono assenti/None. + """ + primary_location = work.get("primary_location") or {} + source = primary_location.get("source") or {} + return source.get("issn_l") or "" + + +def format_so_column(work): + """ + Colonna SO (Journal/Source). Deriva da + `work["primary_location"]["source"]["display_name"]`, con accesso robusto ai + campi annidati: `primary_location` e/o `source` possono essere None per + interi record (es. work senza location primaria), nel qual caso vengono + trattati come dict vuoti invece di sollevare AttributeError/KeyError. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: nome della rivista/sorgente, "" se `primary_location` o `source` + sono assenti/None. + """ + primary_location = work.get("primary_location") or {} + source = primary_location.get("source") or {} + return source.get("display_name") or "" + + +def format_tc_column(work): + """ + Colonna TC (Times Cited). Deriva da `work["cited_by_count"]`, con cast + esplicito a int (a differenza di format_tc_column in format_functions.py, + dove il valore WoS restava una stringa quando presente e un int 0 come + default, generando un tipo misto nella colonna: qui l'obiettivo e' + restituire sempre int). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + int: `cited_by_count` castato a int, 0 se assente/None/non convertibile. + """ + try: + return int(work.get("cited_by_count")) + except (TypeError, ValueError): + return 0 + + +def format_ti_column(work): + """ + Colonna TI (Title). Deriva da `work["title"]`, con fallback su + `work["display_name"]` se `title` e' vuoto/assente (i due campi coincidono + nella maggior parte dei casi osservati, ma `display_name` e' piu' + costantemente popolato). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: titolo del work, "" se sia `title` che `display_name` sono + assenti/vuoti. + """ + title = work.get("title") + if title: + return title + return work.get("display_name") or "" + + +def format_ut_column(work): + """ + Colonna UT (Publication ID / accession number). Deriva da `work["id"]` + (URI OpenAlex, es. "https://openalex.org/W2101234009"), da cui viene + rimosso il prefisso URL per ottenere l'ID nudo (es. "W2101234009"). + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: ID OpenAlex nudo, "" se `work["id"]` e' assente/None. + """ + raw_id = work.get("id") + if not raw_id: + return "" + return raw_id.rsplit("/", 1)[-1] + + +def format_vl_column(work): + """ + Colonna VL (Volume). Deriva da `work["biblio"]["volume"]`, con accesso + robusto ai campi annidati: se `work["biblio"]` e' assente/None, viene + trattato come dict vuoto invece di sollevare AttributeError/KeyError. + + Args: + work: oggetto "work" OpenAlex grezzo. + + Returns: + str: numero di volume, "" se assente. + """ + biblio = work.get("biblio") or {} + return biblio.get("volume") or "" + + +def build_sr_bridge_frame(records): + """ + Costruisce il DataFrame "ponte" da passare, senza alcuna trasformazione, + a metatagextraction.py::SR(M). + + Non e' un adattamento in senso stretto: le chiavi dei record prodotti da + format_au_column/format_db_column/format_ji_column/format_so_column/ + format_py_column in questo stesso modulo si chiamano gia' AU/DB/JI/SO/PY, + esattamente come le colonne che SR(M) si aspetta. Il "ponte" e' quindi solo + l'atto di raccogliere piu' record gia' mappati in un unico pandas.DataFrame + (SR() e' intrinsecamente un'operazione di collezione, non di riga singola: + vedi il modulo docstring per il perche'). + + Args: + records: list[dict], record gia' prodotti da map_work_to_record (o + comunque contenenti almeno le chiavi AU, DB, JI, SO, PY nello + stesso formato prodotto dalle format_XX_column di questo modulo). + + Returns: + pandas.DataFrame costruito direttamente da records, una riga per record, + senza rinominare alcuna colonna. + """ + return pd.DataFrame(records) + + +def compute_sr_for_records(records): + """ + Calcola SR (e SR_FULL) per un'intera collezione di record gia' mappati, + riusando SENZA MODIFICHE metatagextraction.py::SR(M) β€” la stessa funzione + gia' usata altrove nella codebase per lo stesso scopo (es. couplingmap.py, + get_collaborationnetwork.py, entrambe tramite + `metaTagExtraction(df, "SR")`) β€” invece di riscrivere a mano la formula + "Autore, Anno, Rivista" e la sua deduplicazione. + + Questa funzione va chiamata UNA VOLTA sull'intera lista di record (tipicamente + da etl_pipeline.py::_compute_calculated_fields), MAI record-per-record: SR() + delega la deduplicazione a `Series.duplicated()`, che e' significativa solo + se valutata sull'intera collezione. Per questo, a differenza delle altre 34 + colonne, non esiste (e non deve esistere) una format_sr_column(work) a + livello di singolo record in questo modulo. + + Args: + records: list[dict], record gia' mappati con almeno le chiavi AU, DB, + JI, SO, PY (vedi build_sr_bridge_frame). + + Returns: + list[dict]: nuovi dict (i record in input non vengono mutati), ciascuno + arricchito con le chiavi "SR" e "SR_FULL" calcolate da + metatagextraction.py::SR(M) sull'intera collezione. Lista vuota se + `records` e' vuota (SR() non viene invocata: `M["DB"].iloc[0]` solleverebbe + IndexError su un DataFrame vuoto). + """ + if not records: + return [] + + bridge = build_sr_bridge_frame(records) + bridge = SR(bridge) + + enriched = [] + for original, sr_value, sr_full_value in zip(records, bridge["SR"], bridge["SR_FULL"]): + record = dict(original) + record["SR"] = sr_value + record["SR_FULL"] = sr_full_value + enriched.append(record) + return enriched + + +def _reconstruct_abstract(inverted_index): + """ + Funzione interna: ricostruisce il testo lineare di un abstract a partire dalla + struttura "inverted index" di OpenAlex (dict parola -> lista di posizioni), + riordinando le parole secondo le posizioni indicate. + + Algoritmo: determina la posizione massima presente nell'indice, alloca un + array di quella dimensione+1 (inizialmente tutto "buchi"), assegna ogni + parola a ciascuna delle sue posizioni, poi unisce l'array con uno spazio + scartando i "buchi" rimasti vuoti. + + Robustezza (nessuna eccezione sollevata): + - inverted_index assente/None/vuoto -> "". + - posizioni non intere o negative vengono ignorate silenziosamente. + - posizioni duplicate (due parole diverse rivendicano la stessa posizione, + anomalia che non dovrebbe verificarsi in un indice invertito valido ma + viene comunque gestita): vince l'ultima parola incontrata nell'ordine di + iterazione del dict, senza sollevare errori. + - "buchi" nelle posizioni (nessuna parola assegnata a una data posizione, + es. per omissioni nell'indice restituito da OpenAlex): la posizione viene + semplicemente saltata in fase di join, non riempita con placeholder. + + Usata da format_ab_column. + + Args: + inverted_index: dict[str, list[int]] | None, tipicamente + `work["abstract_inverted_index"]`. + + Returns: + str: testo dell'abstract ricostruito, stringa vuota se inverted_index e' + None, vuoto, o non contiene alcuna posizione valida. + """ + if not inverted_index: + return "" + + max_position = -1 + for positions in inverted_index.values(): + for position in positions or []: + if isinstance(position, int) and position > max_position: + max_position = position + + if max_position < 0: + return "" + + slots = [None] * (max_position + 1) + for word, positions in inverted_index.items(): + for position in positions or []: + if isinstance(position, int) and 0 <= position <= max_position: + slots[position] = word + + return " ".join(word for word in slots if word is not None) diff --git a/www/services/type_contracts.py b/www/services/type_contracts.py new file mode 100644 index 000000000..18f5d308e --- /dev/null +++ b/www/services/type_contracts.py @@ -0,0 +1,343 @@ +""" +Schema tipizzato delle 34 colonne bibliometrix-style e funzioni di validazione. + +Questo modulo e' la fonte di verita' sui tipi attesi per ciascuna colonna prodotta +dalla pipeline ETL. In pratica, oggi, l'unico chiamante pianificato e' +etl_pipeline.py sull'output di openalex_mapper.py::map_work_to_record + +compute_sr_for_records (vedi COLUMN_SPECS piu' sotto per una nota importante +sulla scelta di tipo per EM/FU/OA/OI/SC, dove la pipeline WoS storica in +format_functions.py diverge dal nostro mapper OpenAlex). + +Convenzione di tipo: +- campi multi-valore (uno o piu' elementi per pubblicazione) -> list[str] +- campi scalari testuali -> str +- campi scalari numerici -> int + +La lista dei nomi di colonna canonici e' gia' definita in www/services/utils.py +(variabile `columns`); questo modulo la arricchisce con l'informazione di tipo +associata a ciascuna colonna, senza duplicarne la definizione. +""" + +from .utils import * +from enum import Enum + + +class ColumnType(Enum): + """Tipi di dato ammessi per una colonna dello schema a 34 colonne.""" + STRING = "string" + INTEGER = "integer" + STRING_LIST = "string_list" + + +class SchemaValidationError(Exception): + """Sollevata quando un record o un DataFrame non rispetta lo schema atteso + (colonna mancante, tipo non coerente, colonna non riconosciuta in modalita' strict). + Non sollevata direttamente da validate_record/validate_dataframe (che + riportano errori come lista di stringhe, senza sollevare eccezioni): resta + disponibile per un chiamante (es. etl_pipeline.py) che voglia trasformare + una lista di errori di validazione in un'eccezione bloccante.""" + pass + + +# Specifica dichiarativa per ciascuna delle 34 colonne dello schema bibliometrix-style +# (i nomi devono restare allineati a `columns` in www/services/utils.py). +# +# NOTA su EM, FU, OA, OI, SC: nella pipeline WoS storica (format_functions.py) +# questi campi sono inizializzati come liste (es. `emails = []`, `open_access = []`) +# e possono restare multi-valore per WoS/.txt. Nel nostro mapper OpenAlex +# (openalex_mapper.py) restituiscono invece sempre una stringa scalare (EM/FU/OI/SC +# sono sempre "" per decisione esplicita; OA e' un pass-through scalare di +# `oa_status`). Decisione presa in conversazione: qui sono classificati STRING, +# allineati a cio' che produce davvero openalex_mapper.py (l'unico chiamante +# attuale di questo modulo). Se in futuro type_contracts.py dovesse validare +# anche l'output della pipeline WoS storica, questa scelta andra' rivista +# insieme a format_functions.py, non silenziosamente. +# +# "required" qui significa "la chiave deve essere presente nel record" β€” non +# "il valore non puo' essere vuoto": "" / [] / 0 sono rappresentazioni valide +# di "nessun dato", None/NaN non lo sono mai. +# +# BUG DOCUMENTATO E RISOLTO (rilevante per la relazione finale, sezione "Weak +# or inconsistent type enforcement"): PY era inizialmente classificato STRING +# perche' openalex_mapper.py::format_py_column restituiva str(publication_year). +# Questo replicava il tipo GREZZO prodotto dalla pipeline WoS storica +# (format_functions.py::format_py_column restituisce anch'essa una stringa), +# ma non replicava il suo effetto pratico: get_data.py costruisce il +# DataFrame storico con `pd.read_json(StringIO(json))`, che converte +# automaticamente le stringhe numeriche in int64, mentre +# etl_pipeline.py::_build_dataframe usa `pd.DataFrame(records)` diretto, che +# NON fa questa inferenza β€” quindi PY restava str a valle SOLO nella nostra +# pipeline, mai in quella storica. Il sintomo: functions/get_annualproduction.py +# andava in TypeError su `range(min_year, max_year + 1)` perche' min_year/ +# max_year erano stringhe. Verificato con grep su format_functions.py e su +# tutte le functions/*.py che nessun consumer richiede PY come stringa (nessuno +# slicing, nessun accessor .str, nessuna concatenazione); numerosi consumer lo +# richiedono esplicitamente numerico (min/max, range, confronti aritmetici, +# groupby, np.linspace/pd.cut), e 4 di essi (get_authorlocalimpact.py, +# get_authorproductionovertime.py, get_sourceslocalimpact.py, +# get_thematicevolution.py) fanno gia' un cast difensivo +# `pd.to_numeric(..., errors="coerce")` proprio per questo motivo. Risolto +# classificando PY come INTEGER qui e facendo restituire un int nativo da +# format_py_column (vedi openalex_mapper.py per il dettaglio completo). +COLUMN_SPECS: dict = { + "AB": {"type": ColumnType.STRING, "required": True, "description": "Abstract"}, + "AF": {"type": ColumnType.STRING_LIST, "required": True, "description": "Authors Full Name"}, + "AU": {"type": ColumnType.STRING_LIST, "required": True, "description": "Authors"}, + "AU1_UN": {"type": ColumnType.STRING, "required": True, "description": "First Author University"}, + "AU_UN": {"type": ColumnType.STRING_LIST, "required": True, "description": "Authors University"}, + "BP": {"type": ColumnType.STRING, "required": True, "description": "Begin Page"}, + "C1": {"type": ColumnType.STRING_LIST, "required": True, "description": "Authors Affiliations"}, + "CR": {"type": ColumnType.STRING_LIST, "required": True, "description": "Cited References"}, + "DB": {"type": ColumnType.STRING, "required": True, "description": "Source"}, + "DE": {"type": ColumnType.STRING_LIST, "required": True, "description": "Keywords"}, + "DI": {"type": ColumnType.STRING, "required": True, "description": "DOI"}, + "DT": {"type": ColumnType.STRING, "required": True, "description": "Document Type"}, + "EM": {"type": ColumnType.STRING, "required": True, "description": "Author Email"}, + "EP": {"type": ColumnType.STRING, "required": True, "description": "End Page"}, + "FU": {"type": ColumnType.STRING, "required": True, "description": "Funding Details"}, + "FX": {"type": ColumnType.STRING, "required": True, "description": "Acknowledgements"}, + "ID": {"type": ColumnType.STRING_LIST, "required": True, "description": "Index Keywords"}, + "IS": {"type": ColumnType.STRING, "required": True, "description": "Issue"}, + "JI": {"type": ColumnType.STRING, "required": True, "description": "Abbreviated Source Title"}, + "LA": {"type": ColumnType.STRING, "required": True, "description": "Language"}, + "OA": {"type": ColumnType.STRING, "required": True, "description": "Open Access"}, + "OI": {"type": ColumnType.STRING, "required": True, "description": "Author's ORCID"}, + "PMID": {"type": ColumnType.STRING, "required": True, "description": "PubMed ID"}, + "PU": {"type": ColumnType.STRING, "required": True, "description": "Publisher"}, + "PY": {"type": ColumnType.INTEGER, "required": True, "description": "Publication Year"}, + "RP": {"type": ColumnType.STRING, "required": True, "description": "Correspondence Address"}, + "SC": {"type": ColumnType.STRING, "required": True, "description": "Fields of Study"}, + "SN": {"type": ColumnType.STRING, "required": True, "description": "ISSN"}, + "SO": {"type": ColumnType.STRING, "required": True, "description": "Journal"}, + "SR": {"type": ColumnType.STRING, "required": True, "description": "Authors, Publication Year and Journal"}, + "TC": {"type": ColumnType.INTEGER, "required": True, "description": "Time Cited"}, + "TI": {"type": ColumnType.STRING, "required": True, "description": "Title"}, + "UT": {"type": ColumnType.STRING, "required": True, "description": "Publication ID"}, + "VL": {"type": ColumnType.STRING, "required": True, "description": "Volume"}, +} + + +def get_expected_type(column): + """ + Restituisce il ColumnType atteso per una colonna dello schema a 34 colonne. + + Args: + column: nome della colonna (es. "AU", "PY", "TC"). + + Returns: + ColumnType corrispondente, secondo COLUMN_SPECS. + + Raises: + KeyError: se la colonna non fa parte dello schema definito in COLUMN_SPECS. + """ + if column not in COLUMN_SPECS: + raise KeyError(f"Colonna non presente nello schema COLUMN_SPECS: {column!r}") + return COLUMN_SPECS[column]["type"] + + +def is_multivalue(column): + """ + Indica se una colonna e' definita come multi-valore (list[str], es. AU, C1, CR, + DE, ID) oppure scalare (str/int, es. PY, TC, TI, SO). + + Args: + column: nome della colonna. + + Returns: + bool. + + Raises: + KeyError: se la colonna non fa parte dello schema definito in COLUMN_SPECS + (propagata da get_expected_type). + """ + return get_expected_type(column) == ColumnType.STRING_LIST + + +def _is_missing_scalar(value): + """ + Funzione interna: indica se un valore scalare va considerato "mancante" nel + senso proibito dal contratto (None, o NaN in stile pandas/numpy). + + Una stringa vuota "" o una lista vuota [] NON sono considerate mancanti: + sono le rappresentazioni valide di "nessun dato" gia' stabilite nel design + di openalex_mapper.py. Non tenta di valutare la "vacuita'" di list/dict/ + tuple/set (per cui il concetto di NaN non ha senso): restituisce False per + quei tipi senza sollevare eccezioni. + + Args: + value: valore da controllare. + + Returns: + bool: True se value e' None o NaN, False altrimenti (incluse liste/dict + di qualunque contenuto). + """ + if isinstance(value, (list, dict, tuple, set)): + return False + try: + return bool(pd.isna(value)) + except (TypeError, ValueError): + return False + + +def coerce_record_types(record): + """ + Tenta una coercizione "best-effort" dei valori di un record verso i tipi + dichiarati in COLUMN_SPECS, permissiva in input ma con output SEMPRE + conforme allo schema (barriera finale anti-NaN/None richiesta dal design): + + - colonna STRING_LIST: None/NaN -> []; list/tuple gia' presente -> list(); + qualunque altro scalare (es. una singola stringa) -> wrappato in lista + di un elemento. + - colonna INTEGER: None/NaN -> 0; altrimenti tentativo di `int(value)` + (funziona anche su stringhe numeriche come "42" o float come 42.0); + se la conversione fallisce (es. stringa non numerica), fallback a 0 + invece di sollevare un'eccezione. + - colonna STRING: None/NaN -> ""; stringa gia' presente -> invariata; + qualunque altro valore (int, float, list residua, ecc.) -> convertito + con `str(value)`. + + Il record restituito contiene ESATTAMENTE le chiavi di COLUMN_SPECS (quindi + di `columns` in utils.py): colonne mancanti nell'input vengono aggiunte con + il default vuoto del loro tipo, colonne extra presenti nell'input ma non + nello schema vengono scartate. Questo rende la funzione una barriera + robusta anche contro record parziali o con chiavi sporche, non solo contro + valori None/NaN sui campi attesi. + + Args: + record: dict rappresentante una singola riga/pubblicazione. Puo' essere + parziale, avere chiavi extra, o contenere None/NaN/tipi sbagliati: + nessuno di questi casi solleva un'eccezione. + + Returns: + dict: nuovo record con esattamente le chiavi di COLUMN_SPECS, ciascuna + con un valore del tipo python atteso. + """ + coerced = {} + + for column, spec in COLUMN_SPECS.items(): + value = record.get(column) if isinstance(record, dict) else None + expected_type = spec["type"] + + if expected_type == ColumnType.STRING_LIST: + if _is_missing_scalar(value): + coerced_value = [] + elif isinstance(value, list): + coerced_value = value + elif isinstance(value, tuple): + coerced_value = list(value) + else: + coerced_value = [value] + + elif expected_type == ColumnType.INTEGER: + if _is_missing_scalar(value): + coerced_value = 0 + else: + try: + coerced_value = int(value) + except (TypeError, ValueError): + coerced_value = 0 + + else: # ColumnType.STRING + if _is_missing_scalar(value): + coerced_value = "" + elif isinstance(value, str): + coerced_value = value + else: + coerced_value = str(value) + + coerced[column] = coerced_value + + return coerced + + +def validate_record(record, strict=True): + """ + Valida un singolo record (dict colonna -> valore) contro COLUMN_SPECS, + SENZA correggerlo (a differenza di coerce_record_types): riporta soltanto + gli errori trovati. + + Verifica, per ogni colonna prevista dallo schema: + - presenza della chiave, se il campo e' marcato come obbligatorio; + - assenza di valori None/NaN residui (distinti da "" / [] / 0, che sono + rappresentazioni valide di "nessun dato"); + - coerenza del tipo python del valore con quanto dichiarato in COLUMN_SPECS + (list per i campi multi-valore, str per gli scalari testuali, int + β€” esplicitamente NON bool β€” per gli scalari numerici); + - assenza di colonne non riconosciute, se strict=True. + + Args: + record: dict rappresentante una singola riga/pubblicazione, con chiavi + attese tra le 34 colonne dello schema. + strict: se True, chiavi extra non presenti in COLUMN_SPECS sono considerate + un errore di validazione; se False, vengono ignorate. + + Returns: + list[str]: messaggi di errore (lista vuota se il record e' valido). Non + solleva eccezioni: un input non-dict produce un singolo messaggio di + errore descrittivo invece di un TypeError. + """ + if not isinstance(record, dict): + return [f"record non Γ¨ un dict: {type(record).__name__}"] + + errors = [] + + for column, spec in COLUMN_SPECS.items(): + if column not in record: + if spec["required"]: + errors.append(f"{column}: colonna obbligatoria mancante") + continue + + value = record[column] + expected_type = spec["type"] + + if expected_type == ColumnType.STRING_LIST: + if not isinstance(value, list): + errors.append( + f"{column}: atteso list (STRING_LIST), trovato {type(value).__name__} ({value!r})" + ) + + elif expected_type == ColumnType.INTEGER: + if _is_missing_scalar(value): + errors.append(f"{column}: valore mancante (None/NaN) su colonna INTEGER") + elif not isinstance(value, int) or isinstance(value, bool): + errors.append( + f"{column}: atteso int (INTEGER), trovato {type(value).__name__} ({value!r})" + ) + + else: # ColumnType.STRING + if _is_missing_scalar(value): + errors.append(f"{column}: valore mancante (None/NaN) su colonna STRING") + elif not isinstance(value, str): + errors.append( + f"{column}: atteso str (STRING), trovato {type(value).__name__} ({value!r})" + ) + + if strict: + extra = set(record.keys()) - set(COLUMN_SPECS.keys()) + if extra: + errors.append(f"colonne non riconosciute nello schema: {sorted(extra)}") + + return errors + + +def validate_dataframe(df, strict=True): + """ + Valida un intero pandas.DataFrame contro COLUMN_SPECS, applicando + validate_record ad ogni riga e aggregando gli errori con riferimento + all'indice di riga originale del DataFrame (non alla posizione 0-based). + + Args: + df: pandas.DataFrame da validare, atteso con lo schema a 34 colonne + (o un suo sottoinsieme). + strict: vedi validate_record. + + Returns: + list[str]: messaggi di errore, prefissati con "riga : ..." + (lista vuota se il DataFrame e' valido). Un DataFrame vuoto (0 righe) + restituisce una lista vuota senza errori. + """ + errors = [] + for row_index, record in zip(df.index, df.to_dict(orient="records")): + for error in validate_record(record, strict=strict): + errors.append(f"riga {row_index}: {error}") + return errors