iface_ecosio/flask_company_data.py

from datetime import date
import re

import requests
from flask import Flask, jsonify, render_template, request
from requests import Session
from zeep import Client
from zeep.transports import Transport

app = Flask(__name__)

VIES_WSDL_URL = "https://ec.europa.eu/taxation_customs/vies/services/checkVatService.wsdl"
POLISH_WL_API = "https://wl-api.mf.gov.pl/api/search/nip/{nip}?date={query_date}"


def get_vies_client():
    session = Session()
    transport = Transport(session=session, timeout=20)
    return Client(wsdl=VIES_WSDL_URL, transport=transport)


def normalize_vat(country_code: str, vat_number: str):
    country_code = (country_code or "").strip().upper()
    vat_number = (vat_number or "").strip().replace(" ", "").replace(".", "").replace("-", "")

    if vat_number.upper().startswith(country_code):
        vat_number = vat_number[len(country_code):]

    return country_code, vat_number


def parse_polish_working_address(address: str):
    if not address:
        return "", "", ""

    address = " ".join(address.split())
    match = re.match(r"^(.*?)(\d{2}-\d{3})\s+(.+)$", address)

    if match:
        street = match.group(1).strip(" ,")
        zip_code = match.group(2).strip()
        city = match.group(3).strip()
        return street, zip_code, city

    return address, "", ""


def parse_vies_address(address: str):
    """
    Assumption for VIES address:
    - address is comma-separated or line separated
    - first part = street
    - last part = ZIP + CITY or middlepart is city

    Examples:
      "Main Street 12, Some Region, 1000 Brussels"
      "ul. Prosta 12, Warszawa, 00-838 Warszawa"
      "Ilica 1, Zagreb, HR-10000 Zagreb"
    """
    if not address:
        return "", "", ""
 
    parts = [part.strip() for part in address.split(",") if part.strip()]
    if "\n" in address:
        parts = [part.strip() for part in address.split("\n") if part.strip()]
    
    if not parts:
        return "", "", ""

    street = parts[0]
    last_part = parts[-1]

    # Examples:
    # 1000 Brussels
    # 00-838 Warszawa
    # HR-10000 Zagreb
    match = re.match(r"^([A-Z]{0,2}-?\d[\dA-Z -]*)\s+(.+)$", last_part)
    if match:
        zip_code = match.group(1).strip()
        city = match.group(2).strip()
        return street, zip_code, city

    return street, "", last_part


def check_vies(country_code: str, vat_number: str):
    client = get_vies_client()
    result = client.service.checkVat(countryCode=country_code, vatNumber=vat_number)

    name = (result.name or "").strip()
    address = (result.address or "").strip()

    street, zip_code, city = parse_vies_address(address)

    return {
        "source": "vies",
        "found": bool(result.valid),
        "country": country_code,
        "vat_number": vat_number,
        "name": name if result.valid else "",
        "street": street if result.valid else "",
        "zip": zip_code if result.valid else "",
        "city": city if result.valid else "",
        "raw_address": address if result.valid else "",
    }


def check_polish_wl(nip: str):
    today = date.today().isoformat()
    url = POLISH_WL_API.format(nip=nip, query_date=today)

    response = requests.get(url, headers={"Accept": "application/json"}, timeout=20)
    response.raise_for_status()

    data = response.json()
    result = data.get("result", {})
    subject = result.get("subject")

    if not subject:
        subjects = result.get("subjects") or []
        subject = subjects[0] if subjects else None

    if not subject:
        return {
            "source": "mf.gov.pl",
            "found": False,
            "country": "PL",
            "vat_number": nip,
            "name": "",
            "street": "",
            "zip": "",
            "city": "",
            "raw_address": "",
        }

    name = (subject.get("name") or "").strip()
    working_address = (subject.get("workingAddress")or subject.get("residenceAddress") or "").strip()
    street, zip_code, city = parse_polish_working_address(working_address)

    return {
        "source": "mf.gov.pl",
        "found": True,
        "country": "PL",
        "vat_number": nip,
        "name": name,
        "street": street,
        "zip": zip_code,
        "city": city,
        "raw_address": working_address,
    }


@app.get("/company_data/<country>")
def index(country):
    return render_template("company_data.html", country=country)


@app.get("/api/company_lookup")
def company_lookup():
    country = request.args.get("country", "")
    vat_number = request.args.get("vat_number", "")

    country, vat_number = normalize_vat(country, vat_number)

    if not country or not vat_number:
        return jsonify({"error": "country and vat_number are required"}), 400

    try:
        vies_result = check_vies(country, vat_number)
    except Exception as exc:
        vies_result = {
            "source": "ec.europa.eu",
            "found": False,
            "country": country,
            "vat_number": vat_number,
            "name": "",
            "street": "",
            "zip": "",
            "city": "",
            "raw_address": "",
            "warning": f"ec.europa.eu lookup failed: {exc}",
        }

    if vies_result.get("found"):
        return jsonify(vies_result)

    if country == "PL":
        try:
            polish_result = check_polish_wl(vat_number)
            if "warning" in vies_result:
                polish_result["warning"] = vies_result["warning"]
            return jsonify(polish_result)
        except Exception as exc:
            return jsonify(
                {
                    "source": "mf.gov.pl",
                    "found": False,
                    "country": country,
                    "vat_number": vat_number,
                    "name": "",
                    "street": "",
                    "zip": "",
                    "city": "",
                    "raw_address": "",
                    "warning": vies_result.get("warning"),
                    "fallback_error": f"mf.gov.pl lookup failed: {exc}",
                }
            )

    return jsonify(vies_result)


if __name__ == "__main__":
    app.run(debug=True)