""" DST Alkohol og Tobak Harvester & Builder ---------------------------------------- Henter data for 100+ års forbrug og salg af alkohol og tobak fra Danmarks Statistik: - ALKO2: Salg af alkohol og tobak pr. indbygger (1921-2025) - ALKO6: Salg af alkohol og tobak i mængder (1921-2025) - ALKO4: Historisk serie inkl. grænsehandel og faktisk forbrug (1921-2020) Følger fair-use og optimal udtrækning: - Kontrollerer 'updated'-tidsstempel via DST /v1/tableinfo før download. - Cacher rå BULK/JSON-data i en lokal cachemappe. - Genererer både lang/normaliseret (tidy) og bred (pivot) tidsserie i både CSV og JSON format. """ import os import json import csv import requests from datetime import datetime CACHE_DIR = os.path.join(os.path.dirname(__file__), ".cache") OUTPUT_DIR = os.path.dirname(__file__) DST_API_BASE = "https://api.statbank.dk/v1" TABLES = { "ALKO2": "Salg af alkohol og tobak pr. indbygger", "ALKO6": "Salg af alkohol og tobak", "ALKO4": "Forbrug og salg af alkohol og tobak (historisk m. grænsehandel)" } def get_table_metadata(table_id: str) -> dict: url = f"{DST_API_BASE}/tableinfo/{table_id}?format=JSON" resp = requests.get(url) resp.raise_for_status() return resp.json() def fetch_table_data(table_id: str, metadata: dict) -> str: """Henter rå data fra DST som CSV med kodeværdier for stabil parsing.""" os.makedirs(CACHE_DIR, exist_ok=True) cache_meta_file = os.path.join(CACHE_DIR, f"{table_id}_meta.json") cache_data_file = os.path.join(CACHE_DIR, f"{table_id}_raw.csv") last_updated = metadata.get("updated") # Check cache if os.path.exists(cache_meta_file) and os.path.exists(cache_data_file): try: with open(cache_meta_file, "r", encoding="utf-8") as f: cached_meta = json.load(f) if cached_meta.get("updated") == last_updated: print(f"[{table_id}] Cache er up-to-date (sidst opdateret hos DST: {last_updated}). Indlæser fra lokal cache...") with open(cache_data_file, "r", encoding="utf-8-sig") as f: return f.read() except Exception as e: print(f"[{table_id}] Kunne ikke læse cache: {e}. Genhenter...") print(f"[{table_id}] Henter friske data fra DST API (sidst opdateret hos DST: {last_updated})...") payload = { "table": table_id, "format": "CSV", "valuePresentation": "Code", "variables": [ {"code": var["id"], "values": ["*"]} for var in metadata.get("variables", []) ] } resp = requests.post(f"{DST_API_BASE}/data", json=payload) resp.raise_for_status() resp.encoding = "utf-8-sig" content = resp.text # Gem i cache with open(cache_meta_file, "w", encoding="utf-8") as f: json.dump(metadata, f, ensure_ascii=False, indent=2) with open(cache_data_file, "w", encoding="utf-8-sig") as f: f.write(content) return content def parse_float(val: str): if not val or val.strip() in ("..", "-", "...", ""): return None val_clean = val.strip().replace(".", "").replace(",", ".") try: return float(val_clean) except ValueError: return None def main(): print("=== Starter DST Alkohol & Tobak Data Builder ===") all_series = [] # Rækker til normaliseret/lang tabel type_descriptions = {} # code -> beskrivelse # 1. Behandl ALKO2 (pr. indbygger 1921-2025) meta_alko2 = get_table_metadata("ALKO2") type_map_alko2 = {v["id"]: v["text"] for v in meta_alko2["variables"][0]["values"]} type_descriptions.update(type_map_alko2) raw_alko2 = fetch_table_data("ALKO2", meta_alko2) reader = csv.reader(raw_alko2.splitlines(), delimiter=";") header = next(reader) for row in reader: if len(row) >= 3: code, year, val = row[0].strip(), row[1].strip(), row[2].strip() num_val = parse_float(val) all_series.append({ "kilde_tabel": "ALKO2", "kode": f"ALKO2_{code}", "aar": int(year), "kategori": "Pr. indbygger", "beskrivelse": type_map_alko2.get(code, code), "enhed": "Liter ren alkohol pr. indb." if "016" in code or "017" in code else "Cigaretter pr. indb.", "vaerdi": num_val }) # 2. Behandl ALKO6 (mængder 1921-2025) meta_alko6 = get_table_metadata("ALKO6") type_map_alko6 = {v["id"]: v["text"] for v in meta_alko6["variables"][0]["values"]} type_descriptions.update(type_map_alko6) raw_alko6 = fetch_table_data("ALKO6", meta_alko6) reader = csv.reader(raw_alko6.splitlines(), delimiter=";") header = next(reader) for row in reader: if len(row) >= 3: code, year, val = row[0].strip(), row[1].strip(), row[2].strip() num_val = parse_float(val) desc = type_map_alko6.get(code, code) unit = "1.000 liter" if "(1000 liter)" in desc else ("Mio. stk." if "mio. stk" in desc else "Tons") all_series.append({ "kilde_tabel": "ALKO6", "kode": f"ALKO6_{code}", "aar": int(year), "kategori": "Salg mængder", "beskrivelse": desc, "enhed": unit, "vaerdi": num_val }) # 3. Behandl ALKO4 (Grænsehandel og faktisk forbrug 2000-2020) meta_alko4 = get_table_metadata("ALKO4") type_map_alko4 = {v["id"]: v["text"] for v in meta_alko4["variables"][0]["values"]} raw_alko4 = fetch_table_data("ALKO4", meta_alko4) reader = csv.reader(raw_alko4.splitlines(), delimiter=";") header = next(reader) for row in reader: if len(row) >= 3: code, year, val = row[0].strip(), row[1].strip(), row[2].strip() desc = type_map_alko4.get(code, code) # Medtag kun grænsehandel og faktisk forbrug fra ALKO4 (da ordinært salg allerede er dækket og opdateret i ALKO6) if any(term in desc for term in ["Grænsehandel", "Forbrug af", "Fremstilling af"]): num_val = parse_float(val) cat = "Grænsehandel" if "Grænsehandel" in desc else "Faktisk forbrug (inkl. grænsehandel)" unit = "Mio. liter" if "mio. liter" in desc.lower() else ("Mio. stk." if "mio. stk" in desc.lower() else "Tons") all_series.append({ "kilde_tabel": "ALKO4", "kode": f"ALKO4_{code}", "aar": int(year), "kategori": cat, "beskrivelse": desc, "enhed": unit, "vaerdi": num_val }) # Sorter tidsserie kronologisk efter år, kategori, beskrivelse all_series.sort(key=lambda x: (x["aar"], x["kategori"], x["kode"])) # Gem normaliseret tidsserie (lang) som CSV og JSON lang_csv_path = os.path.join(OUTPUT_DIR, "alkohol_tobak_tidsserie_lang.csv") with open(lang_csv_path, "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=["aar", "kilde_tabel", "kode", "kategori", "beskrivelse", "enhed", "vaerdi"]) writer.writeheader() writer.writerows(all_series) print(f"Skrev {len(all_series)} rækker til {lang_csv_path}") lang_json_path = os.path.join(OUTPUT_DIR, "alkohol_tobak_tidsserie_lang.json") with open(lang_json_path, "w", encoding="utf-8") as f: json.dump(all_series, f, ensure_ascii=False, indent=2) print(f"Skrev normaliseret JSON til {lang_json_path}") # 4. Byg Pivot-tabel (Bred: år som række, nøgletal som kolonner) years = sorted(list(set(row["aar"] for row in all_series))) # Udvalgte nøglekolonner til pivot pivot_cols = [ # Pr. indbygger ("alkohol_pr_indb_over_18_liter", "ALKO2_017"), ("alkohol_pr_indb_alle_liter", "ALKO2_016"), ("cigaretter_pr_indb_over_18_stk", "ALKO2_026"), ("cigaretter_pr_indb_alle_stk", "ALKO2_025"), # Samlet salg i ren alkohol (1.000 liter) ("salg_ren_alkohol_i_alt_1000L", "ALKO6_012"), ("salg_oel_ren_alkohol_1000L", "ALKO6_013"), ("salg_vin_ren_alkohol_1000L", "ALKO6_014"), ("salg_spiritus_ren_alkohol_1000L", "ALKO6_015"), # Mængder salg ("salg_oel_pilsnere_1000L", "ALKO6_00"), ("salg_vin_1000L", "ALKO6_05"), ("salg_spiritus_1000L", "ALKO6_06"), ("salg_cigaretter_mio_stk", "ALKO6_022"), ("salg_cigarer_cigarillos_mio_stk", "ALKO6_023"), ("salg_roegtobak_tons", "ALKO6_024"), # Grænsehandel (mio. liter ren alkohol) ("graensehandel_ren_alkohol_mio_L", "ALKO4_155"), ("graensehandel_oel_ren_alkohol_mio_L", "ALKO4_160"), ("graensehandel_vin_ren_alkohol_mio_L", "ALKO4_165"), ("graensehandel_spiritus_ren_alkohol_mio_L", "ALKO4_170"), ("graensehandel_cigaretter_mio_stk", "ALKO4_180"), ] # Map (year, kode) -> vaerdi lookup = {(r["aar"], r["kode"]): r["vaerdi"] for r in all_series} pivot_data = [] for y in years: row = {"aar": y} for col_name, kode in pivot_cols: row[col_name] = lookup.get((y, kode), None) pivot_data.append(row) pivot_fieldnames = ["aar"] + [c[0] for c in pivot_cols] pivot_csv_path = os.path.join(OUTPUT_DIR, "alkohol_tobak_historisk_1921_2025_bred.csv") with open(pivot_csv_path, "w", encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=pivot_fieldnames) writer.writeheader() writer.writerows(pivot_data) print(f"Skrev {len(pivot_data)} år til pivot CSV {pivot_csv_path}") pivot_json_path = os.path.join(OUTPUT_DIR, "alkohol_tobak_historisk_1921_2025_bred.json") with open(pivot_json_path, "w", encoding="utf-8") as f: json.dump(pivot_data, f, ensure_ascii=False, indent=2) print(f"Skrev pivot JSON til {pivot_json_path}") print("=== Fuldført med succes! ===") if __name__ == "__main__": main()