innovadev's picture
Update app.py
20696dd verified
Raw
History Blame Contribute Delete
7.08 kB
import os
import logging
import requests
from flask import Flask, request, jsonify
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger(__name__)
app = Flask(__name__)
GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY", "")
GOOGLE_ADDRESS_VALIDATION_URL = (
"https://addressvalidation.googleapis.com/v1:validateAddress"
)
def evaluate_conditions(dpv_confirmation, dpv_footnote, possible_next_action, metadata):
"""
Returns (passed: bool, failed_reason: str | None)
Conditions are evaluated in order; first match wins.
"""
business = metadata.get("business", False)
# Condition 1
if dpv_confirmation == "Y" and possible_next_action == "ACCEPT":
return True, None
# Condition 2
if (
dpv_confirmation == "Y"
and possible_next_action == "CONFIRM"
and dpv_footnote in {"AABB", "AAU1", "AABBTA", "AABBRR"}
):
return True, None
# Condition 3
if (
dpv_confirmation == "S"
and possible_next_action in {"ACCEPT", "CONFIRM"}
and dpv_footnote in {"AABB", "AAU1", "AABBTA", "AABBRR", "AABBCC","AAC1R7"}
):
return True, None
# Condition 4
if possible_next_action in {"ACCEPT", "CONFIRM"} and dpv_footnote == "AAM3":
return True, None
# Condition 5
if (
dpv_confirmation == "S"
and possible_next_action == "ACCEPT"
and dpv_footnote == "AAC1"
and business is True
):
return True, None
# Condition 6
if possible_next_action == "ACCEPT" and dpv_footnote == "A1":
return True, None
# Condition 7
if dpv_confirmation == "D" and dpv_footnote == "AAN1" and business is True:
return True, None
# Build failure reason
reasons = []
if dpv_confirmation not in {"Y", "S", "D"}:
reasons.append(
f"dpvConfirmation='{dpv_confirmation}' is not Y/S/D"
)
if possible_next_action not in {"ACCEPT", "CONFIRM"}:
reasons.append(
f"possibleNextAction='{possible_next_action}' is not ACCEPT/CONFIRM"
)
if not reasons:
reasons.append(
f"No valid condition matched "
f"(dpvConfirmation='{dpv_confirmation}', "
f"dpvFootnote='{dpv_footnote}', "
f"possibleNextAction='{possible_next_action}', "
f"business={business})"
)
return False, "; ".join(reasons)
@app.route("/", methods=["GET"])
def index():
return """
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
<title>Address Validator API</title>
<style>
body { font-family: Arial, sans-serif; max-width: 800px; margin: 60px auto; padding: 0 20px; color: #333; }
h1 { color: #1a73e8; }
h2 { margin-top: 40px; }
code, pre { background: #f4f4f4; border-radius: 4px; padding: 2px 6px; font-size: 14px; }
pre { padding: 16px; overflow-x: auto; }
.badge { display: inline-block; background: #1a73e8; color: white; padding: 2px 10px; border-radius: 12px; font-size: 13px; margin-right: 6px; }
.badge.post { background: #34a853; }
</style>
</head>
<body>
<h1>Address Validator API</h1>
<p>Validates US addresses using the Google Address Validation API with USPS CASS data.</p>
<h2>Endpoint</h2>
<p><span class="badge post">POST</span> <code>/validate-address</code></p>
<h2>Request Body</h2>
<pre>{
"addressLines": ["1600 Amphitheatre Parkway, Mountain View, CA 94043"], // required
"regionCode": "US", // optional, default: "US"
"enableUspsCass": true // optional, default: true
}</pre>
<h2>Example (curl)</h2>
<pre>curl -X POST https://vamsisnikky-address-validator.hf.space/validate-address \\
-H "Content-Type: application/json" \\
-d '{"addressLines": ["1600 Amphitheatre Parkway, Mountain View, CA 94043"]}'</pre>
<h2>Response</h2>
<pre>{
"final_result": "Passed" | "Failed",
"failed_reason": null | "reason string",
"validation_details": { ... },
"google_response": { ... }
}</pre>
</body>
</html>
""", 200
@app.route("/health", methods=["GET"])
def health():
return jsonify({"status": "ok"}), 200
@app.route("/validate-address", methods=["POST"])
def validate_address():
logger.info("POST /validate-address - remote_addr=%s", request.remote_addr)
body = request.get_json(force=True, silent=True) or {}
logger.info("Request body: %s", body)
address_lines = body.get("addressLines")
region_code = body.get("regionCode", "US")
enable_usps_cass = body.get("enableUspsCass", True)
if not address_lines or not isinstance(address_lines, list):
logger.warning("Missing or invalid addressLines")
return jsonify({"error": "addressLines (array) is required"}), 400
# --- Call Google Address Validation API ---
google_payload = {
"address": {
"regionCode": region_code,
"addressLines": address_lines,
},
"enableUspsCass": enable_usps_cass,
}
try:
google_response = requests.post(
GOOGLE_ADDRESS_VALIDATION_URL,
params={"key": GOOGLE_API_KEY},
json=google_payload,
timeout=10,
)
google_response.raise_for_status()
except requests.exceptions.HTTPError as exc:
logger.error("Google API HTTP error: %s", exc)
return (
jsonify({"error": "Google API error", "detail": str(exc)}),
502,
)
except requests.exceptions.RequestException as exc:
logger.error("Google API request failed: %s", exc)
return (
jsonify({"error": "Failed to reach Google API", "detail": str(exc)}),
502,
)
google_data = google_response.json()
result = google_data.get("result", {})
# --- Extract fields needed for validation ---
verdict = result.get("verdict", {})
usps_data = result.get("uspsData", {})
metadata = result.get("metadata", {})
dpv_confirmation = usps_data.get("dpvConfirmation", "")
dpv_footnote = usps_data.get("dpvFootnote", "")
possible_next_action = verdict.get("possibleNextAction", "")
# --- Evaluate conditions ---
passed, failed_reason = evaluate_conditions(
dpv_confirmation, dpv_footnote, possible_next_action, metadata
)
# --- Build response ---
response_payload = {
"final_result": "Passed" if passed else "Failed",
"failed_reason": failed_reason,
"validation_details": {
"dpvConfirmation": dpv_confirmation,
"dpvFootnote": dpv_footnote,
"possibleNextAction": possible_next_action,
"metadata": metadata,
"verdict": verdict,
},
"google_response": google_data,
}
logger.info("Result: %s | Reason: %s", response_payload["final_result"], failed_reason)
return jsonify(response_payload), 200
if __name__ == "__main__":
app.run(host="0.0.0.0", port=7860)