# Generatore di metadati DCAT-AP_IT in Python
# Produce RDF/Turtle compatibile con dati.gov.it

from rdflib import Graph, Literal, URIRef, Namespace
from rdflib.namespace import DCAT, DCT, FOAF, RDF, XSD
from datetime import datetime, date

# Namespace italiani
DCATAPIT = Namespace("http://dati.gov.it/onto/dcatapit#")
VCARD = Namespace("http://www.w3.org/2006/vcard/ns#")
SKOS = Namespace("http://www.w3.org/2004/02/skos/core#")

def create_dcat_ap_it_metadata(
    catalog_uri: str,
    dataset_id: str,
    title_it: str,
    description_it: str,
    publisher_name: str,
    publisher_uri: str,
    themes: list,
    license_uri: str,
    distributions: list
) -> str:
    """
    Genera metadati DCAT-AP_IT completi in formato Turtle.
    """
    g = Graph()
    g.bind("dcat", DCAT)
    g.bind("dct", DCT)
    g.bind("foaf", FOAF)
    g.bind("dcatapit", DCATAPIT)

    # Definisci il Dataset
    dataset_uri = URIRef(f"{catalog_uri}/dataset/{dataset_id}")

    g.add((dataset_uri, RDF.type, DCAT.Dataset))
    g.add((dataset_uri, RDF.type, DCATAPIT.Dataset))

    # Metadati obbligatori
    g.add((dataset_uri, DCT.title, Literal(title_it, lang="it")))
    g.add((dataset_uri, DCT.description, Literal(description_it, lang="it")))
    g.add((dataset_uri, DCT.modified, Literal(datetime.utcnow().date().isoformat(), datatype=XSD.date)))
    g.add((dataset_uri, DCT.accrualPeriodicity, URIRef("http://publications.europa.eu/resource/authority/frequency/MONTHLY")))
    g.add((dataset_uri, DCT.license, URIRef(license_uri)))

    # Publisher (obbligatorio in DCAT-AP_IT)
    publisher = URIRef(publisher_uri)
    g.add((publisher, RDF.type, DCATAPIT.Agent))
    g.add((publisher, FOAF.name, Literal(publisher_name, lang="it")))
    g.add((publisher, DCT.identifier, Literal(publisher_uri)))
    g.add((dataset_uri, DCT.publisher, publisher))

    # Temi dal vocabolario EU EUROVOC
    for theme_uri in themes:
        g.add((dataset_uri, DCAT.theme, URIRef(theme_uri)))

    # Distribuzione per ogni formato
    for i, dist in enumerate(distributions):
        dist_uri = URIRef(f"{dataset_uri}/distribution/{i}")
        g.add((dist_uri, RDF.type, DCAT.Distribution))
        g.add((dist_uri, RDF.type, DCATAPIT.Distribution))
        g.add((dist_uri, DCAT.accessURL, URIRef(dist["url"])))
        g.add((dist_uri, DCT.format, URIRef(f"http://publications.europa.eu/resource/authority/file-type/{dist['format']}")))
        g.add((dist_uri, DCT.license, URIRef(license_uri)))
        if "bytes" in dist:
            g.add((dist_uri, DCAT.byteSize, Literal(dist["bytes"], datatype=XSD.decimal)))
        g.add((dataset_uri, DCAT.distribution, dist_uri))

    return g.serialize(format="turtle")

# Esempio di utilizzo
rdf_metadata = create_dcat_ap_it_metadata(
    catalog_uri="https://dati.comune.milano.it",
    dataset_id="qualità-aria-2024",
    title_it="Qualità dell'Aria - Rilevazioni 2024",
    description_it="Dataset con le rilevazioni orarie dei sensori di qualità dell'aria nel Comune di Milano",
    publisher_name="Comune di Milano",
    publisher_uri="http://spcdata.digitpa.gov.it/browse/page/Amministrazione/agid",
    themes=["http://publications.europa.eu/resource/authority/data-theme/ENVI"],
    license_uri="https://creativecommons.org/licenses/by/4.0/",
    distributions=[
        {"url": "https://dati.comune.milano.it/dataset/aria-2024.csv", "format": "CSV"},
        {"url": "https://dati.comune.milano.it/dataset/aria-2024.json", "format": "JSON", "bytes": 45230000},
    ]
)

# Utilizzo delle API CKAN di dati.gov.it
# Le API CKAN sono REST con risposta JSON standardizzata

import httpx
import asyncio
from typing import Optional, List

class CKANClient:
    def __init__(self, base_url: str, api_key: Optional[str] = None):
        self.base_url = base_url.rstrip("/")
        self.headers = {"Content-Type": "application/json"}
        if api_key:
            self.headers["Authorization"] = api_key

    async def search_datasets(
        self,
        query: str,
        filters: Optional[dict] = None,
        rows: int = 20,
        start: int = 0
    ) -> dict:
        """
        Cerca dataset nel catalogo CKAN con filtraggio avanzato.
        L'API usa Solr internamente per la ricerca full-text.
        """
        params = {
            "q": query,
            "rows": rows,
            "start": start,
        }

        # Filtri Solr per raffinamento
        if filters:
            fq_parts = [f"{k}:{v}" for k, v in filters.items()]
            params["fq"] = " AND ".join(fq_parts)

        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/api/3/action/package_search",
                params=params,
                headers=self.headers,
                timeout=30.0
            )
            response.raise_for_status()
            result = response.json()

            if not result.get("success"):
                raise ValueError(f"CKAN API error: {result.get('error')}")

            return {
                "total": result["result"]["count"],
                "datasets": result["result"]["results"],
                "page": start // rows + 1,
                "per_page": rows
            }

    async def get_dataset(self, dataset_id: str) -> dict:
        """Recupera un dataset specifico con tutte le sue distribuzioni."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/api/3/action/package_show",
                params={"id": dataset_id},
                headers=self.headers,
                timeout=30.0
            )
            response.raise_for_status()
            result = response.json()

            if not result.get("success"):
                raise ValueError(f"Dataset not found: {dataset_id}")

            return result["result"]

    async def create_dataset(self, dataset_metadata: dict) -> dict:
        """Pubblica un nuovo dataset (richiede API key con permessi di scrittura)."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/api/3/action/package_create",
                json=dataset_metadata,
                headers=self.headers,
                timeout=60.0
            )
            response.raise_for_status()
            return response.json()["result"]

# Utilizzo pratico
async def main():
    client = CKANClient("https://www.dati.gov.it")

    # Cerca dataset ambientali aggiornati
    results = await client.search_datasets(
        query="qualità aria",
        filters={"res_format": "CSV", "groups": "ambiente"},
        rows=10
    )

    print(f"Trovati {results['total']} dataset")
    for ds in results["datasets"]:
        print(f"- {ds['title']} ({ds['num_resources']} risorse)")

asyncio.run(main())

# FastAPI: REST API per open data con paginazione conforme AgID
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import JSONResponse
from typing import Optional, List
from datetime import date
import math

app = FastAPI(
    title="Open Data API - PA Example",
    description="API per la pubblicazione di dati aperti - conforme Linee Guida AgID",
    version="1.0.0"
)

@app.get("/api/v1/datasets/air-quality",
    summary="Rilevazioni qualità aria",
    tags=["Environmental Data"],
    response_model=dict)
async def get_air_quality(
    # Paginazione standard AgID: page + page_size
    page: int = Query(default=1, ge=1, description="Numero pagina (da 1)"),
    page_size: int = Query(default=100, ge=1, le=1000, description="Elementi per pagina (max 1000)"),
    # Filtraggio
    station_id: Optional[str] = Query(default=None, description="ID stazione di rilevamento"),
    pollutant: Optional[str] = Query(default=None, description="Inquinante (PM2.5, PM10, NO2, O3)"),
    date_from: Optional[date] = Query(default=None, description="Data inizio (ISO 8601)"),
    date_to: Optional[date] = Query(default=None, description="Data fine (ISO 8601)"),
    # Ordinamento
    sort_by: str = Query(default="timestamp", description="Campo di ordinamento"),
    sort_order: str = Query(default="desc", regex="^(asc|desc)$"),
    # Formato output
    format: str = Query(default="json", regex="^(json|csv|geojson)$")
):
    """
    Restituisce le rilevazioni di qualità dell'aria con paginazione e filtraggio.

    Supporta output in JSON, CSV e GeoJSON per compatibilità massima.
    Conforme a DCAT-AP_IT e Linee Guida interoperabilità AgID.
    """
    # Query al database con parametri
    offset = (page - 1) * page_size
    records, total_count = await air_quality_service.get_records(
        station_id=station_id,
        pollutant=pollutant,
        date_from=date_from,
        date_to=date_to,
        sort_by=sort_by,
        sort_order=sort_order,
        limit=page_size,
        offset=offset
    )

    total_pages = math.ceil(total_count / page_size)

    # Risposta con metadati di paginazione (pattern AgID)
    response_body = {
        "data": records,
        "meta": {
            "total_count": total_count,
            "page": page,
            "page_size": page_size,
            "total_pages": total_pages,
            "has_next": page < total_pages,
            "has_prev": page > 1,
        },
        "links": {
            "self": f"/api/v1/datasets/air-quality?page={page}&page_size={page_size}",
            "first": f"/api/v1/datasets/air-quality?page=1&page_size={page_size}",
            "last": f"/api/v1/datasets/air-quality?page={total_pages}&page_size={page_size}",
            "next": f"/api/v1/datasets/air-quality?page={page+1}&page_size={page_size}" if page < total_pages else None,
            "prev": f"/api/v1/datasets/air-quality?page={page-1}&page_size={page_size}" if page > 1 else None,
        },
        "dataset": {
            "id": "aria-qualità-2024",
            "title": "Qualità dell'Aria",
            "license": "CC BY 4.0",
            "publisher": "Comune di Milano",
            "last_updated": "2024-12-01"
        }
    }

    # Content negotiation: JSON vs CSV vs GeoJSON
    if format == "csv":
        return Response(
            content=records_to_csv(records),
            media_type="text/csv",
            headers={"Content-Disposition": "attachment; filename=aria-qualità.csv"}
        )
    elif format == "geojson":
        return Response(
            content=records_to_geojson(records),
            media_type="application/geo+json"
        )

    return JSONResponse(content=response_body)

# Strategia di caching per open data API con Redis
from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
import redis.asyncio as redis
import json
import hashlib
from datetime import timedelta

class OpenDataCacheMiddleware:
    """
    Middleware di caching per API open data.
    Usa Redis come cache layer con TTL basato sulla frequenza di aggiornamento del dataset.
    """

    # TTL per tipo di dataset (in secondi)
    DATASET_TTL = {
        "realtime": 60,        # Dati real-time (qualità aria, traffico)
        "daily": 86400,        # Aggiornamento giornaliero
        "weekly": 604800,      # Aggiornamento settimanale
        "monthly": 2592000,    # Aggiornamento mensile
        "static": 31536000,    # Dati statici (confini amministrativi)
    }

    def __init__(self, app, redis_url: str):
        self.app = app
        self.redis = redis.from_url(redis_url)

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        request = Request(scope, receive)

        # Cache solo GET requests
        if request.method != "GET":
            await self.app(scope, receive, send)
            return

        # Genera cache key da URL + query params (ordinati per consistenza)
        cache_key = self._generate_cache_key(str(request.url))

        # Cerca nella cache
        cached = await self.redis.get(cache_key)
        if cached:
            response_data = json.loads(cached)
            response = Response(
                content=response_data["body"],
                status_code=response_data["status_code"],
                headers={
                    **response_data["headers"],
                    "X-Cache": "HIT",
                    "Cache-Control": "public, max-age=3600"
                }
            )
            await response(scope, receive, send)
            return

        # Esegui la request e intercetta la response
        response_body = []

        async def send_wrapper(message):
            if message["type"] == "http.response.body":
                response_body.append(message.get("body", b""))
            await send(message)

        await self.app(scope, receive, send_wrapper)

        # Salva in cache
        if response_body:
            body = b"".join(response_body)
            await self.redis.setex(
                cache_key,
                self.DATASET_TTL["daily"],  # Default: aggiornamento giornaliero
                json.dumps({"body": body.decode(), "status_code": 200, "headers": {}})
            )

    def _generate_cache_key(self, url: str) -> str:
        """Genera cache key stabile dall'URL."""
        return f"opendata:{hashlib.sha256(url.encode()).hexdigest()[:16]}"

# Data Quality Profiler per dataset PA
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Any

@dataclass
class QualityReport:
    dataset_id: str
    total_rows: int
    total_columns: int
    quality_score: float  # 0-100
    issues: List[dict] = field(default_factory=list)
    column_stats: Dict[str, Any] = field(default_factory=dict)

class DataQualityProfiler:
    """
    Profiler di qualità per dataset open data PA.
    Calcola metriche ISO/IEC 25012 e produce un report strutturato.
    """

    REQUIRED_FIELDS = ["id", "timestamp", "value", "station_code"]

    def profile(self, df: pd.DataFrame, dataset_id: str) -> QualityReport:
        report = QualityReport(
            dataset_id=dataset_id,
            total_rows=len(df),
            total_columns=len(df.columns),
            quality_score=100.0
        )

        # 1. Completezza: campi obbligatori
        for field_name in self.REQUIRED_FIELDS:
            if field_name not in df.columns:
                report.issues.append({
                    "severity": "critical",
                    "dimension": "completeness",
                    "field": field_name,
                    "message": f"Campo obbligatorio '{field_name}' mancante"
                })
                report.quality_score -= 20
            else:
                null_pct = df[field_name].isna().sum() / len(df) * 100
                if null_pct > 5:
                    report.issues.append({
                        "severity": "warning",
                        "dimension": "completeness",
                        "field": field_name,
                        "message": f"{null_pct:.1f}% valori NULL nel campo '{field_name}'",
                        "null_count": int(df[field_name].isna().sum()),
                        "null_percentage": null_pct
                    })
                    report.quality_score -= min(10, null_pct)

        # 2. Consistenza: duplicati
        duplicate_count = df.duplicated().sum()
        if duplicate_count > 0:
            dup_pct = duplicate_count / len(df) * 100
            report.issues.append({
                "severity": "warning",
                "dimension": "consistency",
                "message": f"{duplicate_count} righe duplicate ({dup_pct:.1f}%)",
                "duplicate_count": int(duplicate_count)
            })
            report.quality_score -= min(15, dup_pct * 2)

        # 3. Statistiche per colonna
        for col in df.columns:
            col_stats = {
                "dtype": str(df[col].dtype),
                "null_count": int(df[col].isna().sum()),
                "null_percentage": df[col].isna().sum() / len(df) * 100,
                "unique_count": int(df[col].nunique()),
            }
            if df[col].dtype in [np.float64, np.int64]:
                col_stats.update({
                    "min": float(df[col].min()),
                    "max": float(df[col].max()),
                    "mean": float(df[col].mean()),
                    "median": float(df[col].median()),
                })
            report.column_stats[col] = col_stats

        report.quality_score = max(0.0, report.quality_score)
        return report

# Client robusto per consumare open data PA
import httpx
import asyncio
from typing import AsyncIterator
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class RobustOpenDataClient:
    """
    Client per consumo open data con retry, streaming e validazione.
    """

    def __init__(self, timeout: int = 60):
        self.timeout = timeout

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=60),
        retry=retry_if_exception_type((httpx.HTTPError, httpx.TimeoutException))
    )
    async def fetch_dataset(self, url: str) -> dict:
        """Scarica un dataset con retry automatico in caso di errore."""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(url, follow_redirects=True)
            response.raise_for_status()
            return response.json()

    async def stream_large_csv(self, url: str) -> AsyncIterator[dict]:
        """
        Streama CSV di grandi dimensioni senza caricare tutto in memoria.
        Utile per dataset che possono essere di centinaia di MB.
        """
        import csv
        import io

        async with httpx.AsyncClient(timeout=self.timeout) as client:
            async with client.stream("GET", url) as response:
                response.raise_for_status()

                buffer = ""
                headers = None

                async for chunk in response.aiter_text(chunk_size=8192):
                    buffer += chunk
                    lines = buffer.split("\n")

                    # Mantieni l'ultima linea incompleta nel buffer
                    buffer = lines[-1]
                    complete_lines = lines[:-1]

                    if headers is None and complete_lines:
                        headers = list(csv.reader([complete_lines[0]]))[0]
                        complete_lines = complete_lines[1:]

                    if headers:
                        for line in complete_lines:
                            if line.strip():
                                row = list(csv.reader([line]))[0]
                                if len(row) == len(headers):
                                    yield dict(zip(headers, row))

# Utilizzo: import dati ISTAT da API REST
async def import_istat_data():
    client = RobustOpenDataClient()

    # API ISTAT SDMX REST
    istat_url = "https://esploradati.istat.it/SDMXWS/rest/data/IT1,DCSC_POPRES1_EV,1.0/A.IT.9.0?startPeriod=2020"

    try:
        data = await client.fetch_dataset(istat_url)
        # Normalizza il formato SDMX
        return normalize_sdmx_response(data)
    except httpx.HTTPStatusError as e:
        raise RuntimeError(f"ISTAT API error {e.response.status_code}: {e.response.text}")