first release of comapny-data app

flask_company_data.py

@@ -0,0 +1,211 @@
+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(" ", "")
+
+    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
+    - first part = street
+    - last part = ZIP + 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 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": "polish_wl",
+            "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 "").strip()
+    street, zip_code, city = parse_polish_working_address(working_address)
+
+    return {
+        "source": "polish_wl",
+        "found": True,
+        "country": "PL",
+        "vat_number": nip,
+        "name": name,
+        "street": street,
+        "zip": zip_code,
+        "city": city,
+        "raw_address": working_address,
+    }
+
+
+@app.get("/")
+def index():
+    return render_template("index.html")
+
+
+@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": "vies",
+            "found": False,
+            "country": country,
+            "vat_number": vat_number,
+            "name": "",
+            "street": "",
+            "zip": "",
+            "city": "",
+            "raw_address": "",
+            "warning": f"VIES 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": "polish_wl",
+                    "found": False,
+                    "country": country,
+                    "vat_number": vat_number,
+                    "name": "",
+                    "street": "",
+                    "zip": "",
+                    "city": "",
+                    "raw_address": "",
+                    "warning": vies_result.get("warning"),
+                    "fallback_error": f"Polish WL lookup failed: {exc}",
+                }
+            )
+
+    return jsonify(vies_result)
+
+
+if __name__ == "__main__":
+    app.run(debug=True)