#!/usr/bin/env python3 import http.server import json import os import random import re import socketserver import string import threading import time import urllib.error import urllib.request PORT = int(os.getenv("ADDRESS_TOOL_PORT", "18990")) WEB_DIR = os.path.dirname(os.path.abspath(__file__)) UPSTREAM_API = "https://www.meiguodizhi.com/api/v1/dz" FNG_API_KEY = os.getenv("FNG_API_KEY", "").strip() FNG_BASE_URL = "https://www.fakenamegenerator.com/api/v1" FIRST_NAMES = [ "James", "John", "Robert", "Michael", "William", "David", "Richard", "Joseph", "Thomas", "Charles", "Christopher", "Daniel", "Matthew", "Anthony", "Mark", "Donald", "Steven", "Paul", "Andrew", "Joshua", "Kenneth", "Kevin", "Brian", "George", "Edward", "Ronald", "Timothy", "Jason", "Jeffrey", "Ryan", "Jacob", "Gary", "Nicholas", "Eric", "Jonathan", "Stephen", "Larry", "Justin", "Scott", "Brandon", "Mary", "Patricia", "Jennifer", "Linda", "Elizabeth", "Barbara", "Susan", "Jessica", "Sarah", "Karen", "Lisa", "Nancy", "Betty", "Margaret", "Sandra", "Ashley", "Kimberly", "Emily", "Donna", "Michelle", "Carol", "Amanda", "Melissa", "Deborah", "Stephanie", "Rebecca", "Sharon", "Laura", "Cynthia", "Kathleen" ] LAST_NAMES = [ "Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia", "Miller", "Davis", "Rodriguez", "Martinez", "Hernandez", "Lopez", "Gonzalez", "Wilson", "Anderson", "Thomas", "Taylor", "Moore", "Jackson", "Martin", "Lee", "Perez", "Thompson", "White", "Harris", "Sanchez", "Clark", "Ramirez", "Lewis", "Robinson", "Walker", "Young", "Allen", "King", "Wright", "Scott", "Torres", "Nguyen", "Hill", "Flores" ] US_STATE_CODES = { "alabama": "AL", "alaska": "AK", "arizona": "AZ", "arkansas": "AR", "california": "CA", "colorado": "CO", "connecticut": "CT", "delaware": "DE", "florida": "FL", "georgia": "GA", "hawaii": "HI", "idaho": "ID", "illinois": "IL", "indiana": "IN", "iowa": "IA", "kansas": "KS", "kentucky": "KY", "louisiana": "LA", "maine": "ME", "maryland": "MD", "massachusetts": "MA", "michigan": "MI", "minnesota": "MN", "mississippi": "MS", "missouri": "MO", "montana": "MT", "nebraska": "NE", "nevada": "NV", "new hampshire": "NH", "new jersey": "NJ", "new mexico": "NM", "new york": "NY", "north carolina": "NC", "north dakota": "ND", "ohio": "OH", "oklahoma": "OK", "oregon": "OR", "pennsylvania": "PA", "rhode island": "RI", "south carolina": "SC", "south dakota": "SD", "tennessee": "TN", "texas": "TX", "utah": "UT", "vermont": "VT", "virginia": "VA", "washington": "WA", "west virginia": "WV", "wisconsin": "WI", "wyoming": "WY", "district of columbia": "DC" } MIN_COOLDOWN_SECONDS = 0.5 REQUEST_TIMEOUT = 10 RATE_LIMIT_COOLDOWN = 15 _lock = threading.Lock() _is_requesting = False _last_request_time = 0.0 _cooldown_until = 0.0 def generate_secure_password(length=14): upper = random.choice(string.ascii_uppercase) lower = random.choice(string.ascii_lowercase) digit = random.choice(string.digits) chars = string.ascii_letters + string.digits rest = "".join(random.choices(chars, k=length - 3)) pwd_list = list(upper + lower + digit + rest) random.shuffle(pwd_list) return "".join(pwd_list) def resolve_us_state(city_str): if not city_str: return "" c_lower = city_str.strip().lower() if len(c_lower) == 2 and c_lower.upper() in US_STATE_CODES.values(): return c_lower.upper() for name, code in US_STATE_CODES.items(): if name in c_lower or c_lower == name: return code return "" def transform_fng_to_dz(item): name_obj = item.get("name", {}) addr_obj = item.get("address", {}) card_obj = item.get("credit_card", {}) bday_obj = item.get("birthday", {}) height_obj = item.get("height", {}) weight_obj = item.get("weight", {}) first = name_obj.get("first", "") last = name_obj.get("last", "") full = name_obj.get("full", f"{first} {last}".strip()) # 格式化过期时间 YYYY-MM -> MM/YYYY expires_raw = card_obj.get("expires", "") expires = expires_raw if "-" in expires_raw: parts = expires_raw.split("-") if len(parts) == 2: expires = f"{parts[1]}/{parts[0]}" height_str = "" if height_obj.get("in") and height_obj.get("cm"): height_str = f"{height_obj.get('in')}\" ({height_obj.get('cm')}cm)" weight_str = "" if weight_obj.get("lb") and weight_obj.get("kg"): weight_str = f"{weight_obj.get('lb')} lbs ({weight_obj.get('kg')}kg)" transformed = { "Address": addr_obj.get("street", ""), "Telephone": item.get("phone", {}).get("number", ""), "Address_Alias": "", "Trans_Address": "", "Trans_Cn_Address": "", "Fax": "", "City": addr_obj.get("city", ""), "Zip_Code": addr_obj.get("postal_code", ""), "rowkey": item.get("id", ""), "State": addr_obj.get("state", ""), "State_Full": addr_obj.get("state", ""), "Expires": expires, "Credit_Card_Type": card_obj.get("network", "Visa"), "Credit_Card_Number": card_obj.get("number", ""), "CVV2": str(card_obj.get("cvv", "")), "Full_Name": full, "Gender": (item.get("gender") or "Male").capitalize(), "Full_Name_Tran": "", "Title": name_obj.get("title", "Mr."), "Birthday": bday_obj.get("date", ""), "Username": item.get("username", ""), "Password": item.get("password", ""), "Height": height_str, "Weight": weight_str, "Temporary_mail": item.get("email", ""), "System": "Windows 11", "GUID": item.get("id", ""), "Blood_Type": item.get("blood_type", "O+"), "Browser_User_Agent": item.get("user_agent", ""), "Occupation": item.get("occupation", ""), "Company_Name": item.get("company", ""), "Social_Security_Number": item.get("national_id", {}).get("value", "") if item.get("national_id") else "", "First_Name": first, "Last_Name": last, "Generated_Password": generate_secure_password(14), "_source": "fakenamegenerator.com" } return transformed def fetch_from_fng(city_or_state): url = f"{FNG_BASE_URL}/identities?count=1&country=us" state_code = resolve_us_state(city_or_state) if state_code: url += f"&state={state_code}" req = urllib.request.Request( url, headers={ "Authorization": f"Bearer {FNG_API_KEY}", "User-Agent": "OpenClaw-AddressTool/2.0" } ) with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as response: if response.status == 200: res_data = json.loads(response.read().decode("utf-8")) if res_data.get("data") and len(res_data["data"]) > 0: addr_transformed = transform_fng_to_dz(res_data["data"][0]) return {"status": "ok", "address": addr_transformed} return None class ProxyHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=WEB_DIR, **kwargs) def send_standard_json(self, status_code, data): self.send_response(status_code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode("utf-8")) def do_POST(self): global _is_requesting, _last_request_time, _cooldown_until if self.path == "/api/v1/dz" or self.path == "/api/generate": content_length = int(self.headers.get("Content-Length", 0)) post_data = self.rfile.read(content_length) now = time.time() with _lock: if _is_requesting: self.send_standard_json(429, { "status": "busy", "error_code": "ERR_CONCURRENT_REQUEST", "message": "正在处理上一条生成请求，请稍候", "retry_after": 1 }) return _is_requesting = True try: payload = {} try: payload = json.loads(post_data.decode("utf-8")) except Exception: pass city = payload.get("city", "") # 优先调用官方 FakeNameGenerator API if FNG_API_KEY: try: fng_result = fetch_from_fng(city) if fng_result: self.send_standard_json(200, fng_result) return except Exception as e: print(f"[WARN] FNG API failed, falling back to upstream: {e}") # 降级备用：原有的 meiguodizhi.com 接口 req = urllib.request.Request( UPSTREAM_API, data=post_data, headers={ "Content-Type": "application/json", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36", "Referer": "https://www.meiguodizhi.com/" } ) try: with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as response: raw_body = response.read() data = json.loads(raw_body.decode("utf-8")) if data.get("status") == "ok" and "address" in data: addr = data["address"] orig_fn = addr.get("Full_Name", "") # 格式化 cleaned = re.sub(r"[^a-zA-Z\s]", "", orig_fn).strip() words = [w.capitalize() for w in cleaned.split() if w] if len(words) >= 2: fn = f"{words[0]} {words[-1]}" addr["First_Name"] = words[0] addr["Last_Name"] = words[-1] else: fn = f"{random.choice(FIRST_NAMES)} {random.choice(LAST_NAMES)}" p = fn.split() addr["First_Name"] = p[0] addr["Last_Name"] = p[1] addr["Full_Name"] = fn addr["Generated_Password"] = generate_secure_password(14) addr["_source"] = "meiguodizhi.com" self.send_standard_json(200, data) except Exception as e: self.send_standard_json(500, { "status": "error", "message": f"生成失败: {str(e)}" }) finally: with _lock: _is_requesting = False _last_request_time = time.time() else: self.send_error(404, "Not Found") def do_OPTIONS(self): self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "POST, GET, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() if __name__ == "__main__": socketserver.TCPServer.allow_reuse_address = True with socketserver.TCPServer(("0.0.0.0", PORT), ProxyHandler) as httpd: print(f"Address tool server running on port {PORT}") httpd.serve_forever()