Postin osoitetiedot nuts -> sqlite.db v.0.1

main.py

#!/usr/bin/env python3


# (c) Juho Vähäkangas 2024

# lisensöimätön

# Tämä skripti ei tarvitse virtualenviä.
# käytettävät importit ovat pythonin perus moduuleja.

#
# DOKUMENTTI JA POSTIN KÄYTTÖEHDOT
#
# Äkkiseltään en löytänyt pdf tiedosto kuvausta suomeksi.
# Tosin käytin suomalaista  lähdettä, mutta nyt en löytänyt uudestaan sitä.
# Löytyy englanniksi googlella "nuts3 posti osoite"
# PDF, "SERVICE DESCRIPTION AND TERMS OF USE" 1.1.2015
# Tiedostossa on kyseessä pitkistä rivistä, joissa alkaa tietystä merkkikohdasta ja on x merkkiä pitkä
# kohta, josta saadaan tietty osa luettua esim. kadun nimi.

# Kyseessä on NUTS3 data tiedostojen purku tietokantaan
# Tietkonata rakentuu seuraavasti

# Postin osoitetieto on noin 100 M
# sqlite:stä tuleva tiedoston koko on noin 16 M

#
# TIETOKANTA JA SEN LYHYT KUVAUS
#
# address:
# | ID | address

# id toimii pointerina, johon seuraavan taulun address_id osoittaa
# address on unique tyypinen

# loc: 
# | even_odd | min | max | zip | city | address_id

# Tässä demossa ei ole hyödynnetty parittoman tarkistusta, mutta lyhykäisyydessä:
# 1 = pariton, 2 = parillinen.
# tietokanta luodaan tosin että pariton on false ja parillinen true ja null jos ei ole tietoa
# min - max ovat numerot joiden väli osuu kyseiseen postinumeroon
# city on kaupungin / kunnan nimi
# address_id on osoitus toiseen tauluu, josta saadaan kadun nimi 

#
# HUOMIOT 
#
# Tästä puuttuu myös ruotsinkielisten paikkojen hyödyntäminen, toimii siis suomenkielille.
#
# Huomasin myös, että osa nimistä löytyy vain ruotsinkielisinä. Lienee myös toisinpäin, joten
# jos haluaa jatkokehitellä, niin tästä jatkoa, että  jos ei löydy suomeksi niin sitten
# käytetään ruotsiksi.

# Init jaettu osiin kun huomasin, että osoitetietojen jälkeen oli häiriöitä jälkimmäisen luupin kanssa
# Johtunee siitä että ohjelma lukee muistiin koko nuts tiedoston, jossa on 407290 riviä (16.11.24) 
# init1: Ajaa "brute forcella" kadun nimiä tietokantaan, koska tietotyyppi on unique
# niin tieto menee vain kerran.
# init2: Tekee rankemman työn ja säätää yhteenkuuluvuudet

# Ajot kannattaa tehdä tehokkaalla koneella.




import sys
import os
import sqlite3

import re


import datetime
import zipfile
import json


from urllib.request import urlopen
from urllib.parse import urlparse, unquote_plus, unquote
from http.server import HTTPServer, BaseHTTPRequestHandler


temp_data = "temp.zip"
dbname = "osoitteet.db"


schema_sql = """
CREATE TABLE address (
	id INT PRIMARY_KEY NOT NULL,
	address TEXT NOT NULL,
	UNIQUE(address)
);

CREATE TABLE loc (
    even_odd BOOLEAN,
	min INT,
	max INT,
	zip TEXT,
	city TEXT,
	address_id INTEGER NOT NULL,
	FOREIGN KEY (address_id)
       		REFERENCES address (id)	
);


"""


class Downloader():
    def __init__(self):
        if not os.path.isfile(temp_data):
            self.fetch()
            self.unzip()
        if not os.path.isfile(self.filename(purged=True)):
            self.unzip()
    def last_saturday(self):
        today = datetime.date.today()
        nextSaturday = today + datetime.timedelta(days=-today.weekday()+5, weeks=-1)
        return nextSaturday.strftime("%Y%m%d")
    def filename(self, purged=False):
        if purged:
            data = "BAF_" + self.last_saturday() + ".dat"
            return data
        else:
            return "BAF_" + self.last_saturday() + ".zip"
    def fetch(self):
        if os.path.isfile(temp_data):
            print("Vanha temp.zip löytyi. Käytetään sitä.")
            return
        print("Ladataan viime lauantain zip.")
        content = urlopen(f"https://www.posti.fi/webpcode/{self.filename()}").read()
        with open(temp_data, "wb+") as f:
            f.write(content)
            f.close()
    def unzip(self):
        print("Puretaan zip.")
        with zipfile.ZipFile(temp_data, "r") as zip_ref:
            zip_ref.extractall(".")
    def clean(self):
        os.remove(temp_data)
        return



#
# init scripti asettaa nimet tietokantaan
# Skipataan seuraavat:
# - tyhjät rivit
# - PL <numero> rivit

# TODO:
# - poistetaan SMARTPOST ja muuta ei tarpeellista

class Database:

    def __init__(self, dbname):
        self.conn = sqlite3.connect(dbname)

    def __enter__(self):
        self.conn = self.conn.__enter__()
        self.curs = self.conn.cursor()
        return self.curs

    def __exit__(self, *exc_info):
        result = self.conn.__exit__(*exc_info)
        self.curs.close()
        self.conn.close()
        return result
    def initialize(self, step=1):
        num = 0
        try:
            print("Luodaan sql tietokanta")
            with Database(dbname) as conn:
                conn.executescript(schema_sql)
        except sqlite3.OperationalError:
            print("Tietokanta on jo luotu. ")
        print("Luetaan postin data tiedosto. Voi kestää hetken, riippuen koneen tehoista.")
        posti_data = Downloader().filename(purged=True)
        file = open(posti_data, "r", encoding="latin-1")
        
        #
        # Laitetaan kadun nimet tietokantaan
        #
        if step == 1:
            print("Laitetaan kadun nimet tietokantaan...", end=" ")

            nuts = Nuts()
            while line := file.readline():
                num += 1
                address = nuts.parse_address(line)
                if address == None:
                    continue
                try: 
                    with Database(dbname) as conn:
                        conn.execute("INSERT INTO address VALUES (?, ?)", (num, address,))
                except sqlite3.IntegrityError:
                    pass
            print("valmis.")
            
        # 
        # Laitetaan postinumerot ja katunumerot tietokantaan
        #
        if step == 2:
            print("Laitetaan postinumerot tietokantaan...", end=" ")
            nuts = Nuts()
            while line := file.readline():
                address  = nuts.parse_address(line)
                if address == None:
                    continue

                zipcode  = line[13:(13+5)]
                city     = nuts.parse_city(line)
                even_odd = nuts.parse_even_odd(line) 
                num_min  = nuts.to_int(line[187:(187+6)])
                num_max  = nuts.to_int(line[200:(200+6)])
                try:
                    with Database(dbname) as conn:
                        c = conn.execute("SELECT id, address FROM address WHERE address = ?", (address,))
                        current_address = c.fetchone()
                        conn.execute("INSERT INTO loc (even_odd, min, max, zip, city, address_id) VALUES (?, ?, ?, ?, ?, ?)", (even_odd, num_min, num_max, zipcode, city, current_address[0],)) 
                        
                except sqlite3.IntegrityError as ie:
                    print(ie)
            print("valmis.")
    def search(self, q):
        
        address_number = 0
        address_name = ""
        result = []
        simple = False

        if re.findall('^[0-9]+', q): 
            query = re.search(r'(^[0-9]+)\s+(\w+)', q)
            if len(query.group()) > 0:
                address_number = query.group(1)
                address_name = query.group(2)    
        else:
            simple = True
        with Database(dbname) as conn:
            if simple:
                query = q
            else:
                query = address_name
            streets = conn.execute("SELECT id, address FROM address WHERE address LIKE ? LIMIT 10", ('%'+ query +'%',))
            if simple:
                result = []
                for street in streets.fetchall():
                    sql = "SELECT zip, city, address_id FROM loc WHERE address_id = ? LIMIT 10"
                    zips = conn.execute(sql, (street[0],))
                    for z in zips:
                        r = {
                            "address" : street[1],
                            "zip": z[0],
                            "city": z[1],
                            "number": address_number,
                        }
                        result.append(r)
                return result
            for street in streets.fetchall():
                sql = "SELECT zip, city, address_id FROM loc WHERE (? BETWEEN min AND max) AND address_id = ? LIMIT 2"
                zips = conn.execute(sql, (int(address_number), street[0],))

                for z in zips:
                    r = { "address": street[1],
                       "zip": z[0],
                       "city": z[1],
                       "number": address_number}
                    result.append(r)
        return result



class Nuts:
    def parse_address(self, line):
        line = line[102:(103+29)]
        if "PL" in line:
            return 
        if line.startswith(" "):
            return 
        return line.strip()

    def parse_city(self, line):
        line = line[216:(216+20)]
        return line.strip()
    
    def parse_even_odd(self, line):
        line = line[186:(187+1)]
        if '2' in line:
            return True
        if '1' in line:
            return False
        return None

    def to_int(self, str_int):
        try:
            return int(str_int.strip(" "))
        except:
            return None
        
    

class Server(BaseHTTPRequestHandler):
    def _set_header(self):
        self.send_response(200)
        self.send_header("Content-type", "text/json")
        length = int(self.headers['Content-Length'])
        content = self.rfile.read(length)
        temp = str(content).strip('b\'')
        self.end_headers()
        return temp

    def do_POST(self):
        self.send_response(200)
        self.send_header('Content-type', 'text/json')
        self.data_string = self.rfile.read(int(self.headers['Content-Length']))
        self.end_headers()
        data = json.loads(self.data_string)
        db = Database(dbname)
        search_result = db.search(data["q"])
        db.conn.close()
        self.wfile.write(json.dumps(search_result).encode('utf-8'))
        return

def clean():
    pass
"""
    print("Poistetaan purettu tiedosto.")
    os.remove(Downloader().filename(purged=False))
    print("Poistetaan zip tiedosto.")
    os.remove(Downloader().filename(purged=True))
    print("Poistetaan temp.zip tiedosto")
    os.remove("temp.zip")
"""
def usage():
        print("""
              
=============== Kuvaus =============== 

    Demo v 0.1              
    (c) Juho Vähäkangas

    Tämä on demo versio suomessa olevien osoitteiden hakuun.
    Mitä ohjelma tekee?
              
    Ohjelmalla voit 

        o Lataa postin palvelimelta uuden tiedoston, joka sisältää suomessa olevat katuosoitteet

        o Lataa tekstitiedosto sqlite3 tietokantaan
              
        o Pitää pienimuotoista api serveriä

    Vinkki vitonen: (molemmissa vinkeissä joudut puukottamaan .py tiedostoa)

        o Jos haluat käyttää apia muulta koneelta muuta 127.0.0.1 => 0.0.0.0

        o Jos portti 8888 on varattu laita jokin vapaa portti tilalle


=============== Ohjelma ===============

main.py <komento>
              

Komennot ovat:
    download
        Lataa uusimman osoitetiedoston, purkaa sen.
              
    init1
        Lataa tekstitiedoston sqlite tietokantaan (katuosoitteet).
              
    init2
        Lataa postinumerot ja muut tärkeät tiedot tietokantaan (lataa ensiksi init1)
              
    clean
        Poistaa turhat tiedostot jos kaikki on tehty (eli jos serve toimii)
              
    serve
        Pitää päällä pienimuotoista api palvelinta, jolle voi lähettää komentoja.
        Esimerkki komennosta: 
        
        curl -H "Content-Type: application/json" -X POST -d '{"q" : "20 Mannerh"}' "localhost:8888/"
        curl -H "Content-Type: application/json" -X POST -d '{"q" : "Mannerh"}'  "localhost:8888/"
        

        Ideana käyttää javascriptillä esim. laskutusohjelmassa.
              """)

if __name__ == "__main__":
    if len(sys.argv) == 1:
        usage()
        sys.exit(0)
    if sys.argv[1] == "help":
        usage()
    if sys.argv[1] == "download":
        Downloader()
    if sys.argv[1] == "init1":
        Database.initialize(dbname, step=1)
    if sys.argv[1] == "init2":
        Database.initialize(dbname, step=2)
    if sys.argv[1] == "serve":
        with HTTPServer(('0.0.0.0', 8889), Server) as httpd:
            httpd.serve_forever()

Oho, nähtävästi jäikin että porttina on 8889 ja bindaa 0.0.0.0, mutta jos haluat koneen sisäisesti niin vaihda 127.0.0.1 (eli tällöin ulkopuolelta ei pääse api serveriin käsiksi)

https://www.posti.fi/mzj3zpe8qb7p/1eKbwM2WAEY5AuGi5TrSZ7/33cfc2c66d2649af885b36e3935556a1/posti-postinumeropalvelut-palvelukuvaus-ja-kayttoehdot-20150101.pdf