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 """
Validates US addresses using the Google Address Validation API with USPS CASS data.
POST /validate-address
{
"addressLines": ["1600 Amphitheatre Parkway, Mountain View, CA 94043"], // required
"regionCode": "US", // optional, default: "US"
"enableUspsCass": true // optional, default: true
}
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"]}'
{
"final_result": "Passed" | "Failed",
"failed_reason": null | "reason string",
"validation_details": { ... },
"google_response": { ... }
}
""", 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)