279 lines
12 KiB
Python
279 lines
12 KiB
Python
#!/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()
|