Initial commit: Unified API Gateway v3.0.0

This commit is contained in:
dvs
2026-08-28 10:06:18 +08:00
commit 7d9e221993
5 changed files with 3388 additions and 0 deletions
+578
View File
@@ -0,0 +1,578 @@
"""
Unified API Gateway v3.0
Wraps all external APIs in /api/v3/ format
Supports precise city name matching (adcode.txt)
Server deployment version
"""
import json
import re
import requests
from flask import Flask, request, jsonify
from datetime import datetime
app = Flask(__name__)
# ========== Configuration ==========
# Time API - using api.zxki.cn (simple and reliable)
TIME_API_URL = "https://api.zxki.cn/api/time"
# Weather API - sojson (uses citykey)
WEATHER_SOJSON_URL = "http://t.weather.sojson.com/api/weather/city/{}"
# Weather API - uapis.cn (general weather interface)
WEATHER_UAPIS_URL = "https://uapis.cn/api/v1/misc/weather"
# Phone number lookup API
PHONE_INFO_URL = "https://uapis.cn/api/v1/misc/phoneinfo"
# IP Geolocation APIs
IP_MYIP_URL = "https://uapis.cn/api/v1/network/myip"
IP_INFO_URL = "https://uapis.cn/api/v1/network/ipinfo"
IP_SB_URL = "https://api.ip.sb/geoip/{}"
# ========== City Code Mapping ==========
CITY_CODE_MAP = {}
def load_city_codes():
"""Load city code mapping from adcode.txt"""
global CITY_CODE_MAP
try:
with open('adcode.txt', 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line and '=' in line:
code, city = line.split('=', 1)
city_name = city.strip()
code_value = code.strip()
CITY_CODE_MAP[city_name] = code_value
print(f"Loaded {len(CITY_CODE_MAP)} city codes")
except FileNotFoundError:
print("Error: adcode.txt not found, city lookup unavailable")
raise SystemExit("Missing adcode.txt file")
except Exception as e:
print(f"Error loading adcode.txt: {e}")
raise
def search_city_code(query):
"""
Search for city code
1. Remove city/county/district suffixes and search
2. If multiple matches found, search again with suffix
3. If still multiple matches, return error
4. Single-character search not supported
"""
if not query or len(query.strip()) == 0:
return None, "City name cannot be empty"
query = query.strip()
# Single-character search not supported
if len(query) <= 1:
return None, "Single-character search not supported, please enter full city name"
# Remove suffix
clean_query = re.sub(r'[市县区州]$', '', query)
# First search: exact match on name without suffix
if clean_query in CITY_CODE_MAP:
return CITY_CODE_MAP[clean_query], None
# If suffix was removed and original query exists, try original
if clean_query != query and query in CITY_CODE_MAP:
return CITY_CODE_MAP[query], None
# Search for partial matches
matches = {}
for city, code in CITY_CODE_MAP.items():
# City name equals query without suffix
if city == clean_query:
matches[city] = code
# Query without suffix is a prefix of city name
elif city.startswith(clean_query):
# Require city to be at least 1 char longer than query to avoid vague matches
if len(city) - len(clean_query) >= 1:
matches[city] = code
# If first search found results
if matches:
if len(matches) == 1:
city, code = list(matches.items())[0]
return code, None
else:
# Multiple results, try adding original suffix back
if clean_query != query:
# Search for city names containing original query
refined_matches = {}
for city, code in CITY_CODE_MAP.items():
if query in city:
refined_matches[city] = code
if len(refined_matches) == 1:
city, code = list(refined_matches.items())[0]
return code, None
elif len(refined_matches) > 1:
cities = list(refined_matches.keys())
return None, f"Multiple cities found: {', '.join(cities)}, please provide a more specific city name"
else:
cities = list(matches.keys())
return None, f"Multiple cities found: {', '.join(cities)}, please provide a more specific city name"
else:
cities = list(matches.keys())
return None, f"Multiple cities found: {', '.join(cities)}, please provide a more specific city name"
# No match found
return None, f"City '{query}' not found"
# ========== Generic Request Function ==========
def safe_request(url, method='GET', params=None, timeout=10):
"""Safe HTTP request handler"""
try:
if method == 'GET':
resp = requests.get(url, params=params, timeout=timeout)
elif method == 'POST':
resp = requests.post(url, json=params, timeout=timeout)
else:
return None
if resp.status_code == 200:
return resp.json()
elif resp.status_code == 404:
return {"error": "not_found", "message": "Resource not found"}
else:
return {"error": f"HTTP {resp.status_code}", "message": resp.text}
except requests.exceptions.Timeout:
return {"error": "timeout", "message": "Request timeout"}
except requests.exceptions.ConnectionError:
return {"error": "connection_error", "message": "Connection failed"}
except requests.RequestException as e:
return {"error": "request_failed", "message": str(e)}
except json.JSONDecodeError:
return {"error": "json_decode_failed", "message": "Response parsing failed"}
# ========== Response Formatting ==========
def success_response(data, message="Success"):
"""Unified success response format"""
return jsonify({
"code": 200,
"message": message,
"data": data
})
def error_response(code, message, http_status=400):
"""Unified error response format"""
return jsonify({
"code": code,
"message": message
}), http_status
# ========== Helper: Get Local Time ==========
def get_local_time():
"""Get local time as fallback when external API fails"""
now = datetime.now()
weekdays_cn = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"]
weekdays_short = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
weekdays_en = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
return {
"timestamp": str(int(now.timestamp())),
"timestamp_ms": str(int(now.timestamp() * 1000)),
"datetime": now.strftime("%Y-%m-%d %H:%M:%S"),
"date_cn": now.strftime("%Y年%m月%d日"),
"date": now.strftime("%Y年%m月%d日"),
"time": now.strftime("%H:%M:%S"),
"week_num": str(now.isocalendar()[1]),
"week_cn": weekdays_cn[now.weekday()],
"week_short_cn": weekdays_short[now.weekday()],
"week_en": weekdays_en[now.weekday()]
}
# ========== API Routes ==========
@app.route('/api/v3/time', methods=['GET'])
def get_time():
"""
Get current time
GET /api/v3/time
"""
result = safe_request(TIME_API_URL)
if result and "date" in result and "time" in result:
try:
date_str = result.get("date", "")
time_str = result.get("time", "")
week_str = result.get("week", "")
try:
dt = datetime.strptime(f"{date_str} {time_str}", "%Y年%m月%d日 %H:%M:%S")
timestamp = int(dt.timestamp())
timestamp_ms = int(dt.timestamp() * 1000)
datetime_iso = dt.strftime("%Y-%m-%d %H:%M:%S")
except ValueError:
now = datetime.now()
try:
year = int(re.search(r'(\d{4})年', date_str).group(1))
month = int(re.search(r'年(\d{2})月', date_str).group(1))
day = int(re.search(r'月(\d{2})日', date_str).group(1))
hour, minute, second = map(int, time_str.split(':'))
dt = datetime(year, month, day, hour, minute, second)
timestamp = int(dt.timestamp())
timestamp_ms = int(dt.timestamp() * 1000)
datetime_iso = dt.strftime("%Y-%m-%d %H:%M:%S")
except:
now = datetime.now()
timestamp = int(now.timestamp())
timestamp_ms = int(now.timestamp() * 1000)
datetime_iso = now.strftime("%Y-%m-%d %H:%M:%S")
weekdays_cn = ["星期一", "星期二", "星期三", "星期四", "星期五", "星期六", "星期日"]
weekdays_short = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
weekdays_en = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"]
weekday_index = 0
if week_str in weekdays_cn:
weekday_index = weekdays_cn.index(week_str)
else:
try:
weekday_index = dt.weekday()
except:
weekday_index = datetime.now().weekday()
formatted_data = {
"timestamp": str(timestamp),
"timestamp_ms": str(timestamp_ms),
"datetime": datetime_iso,
"date_cn": date_str,
"date": date_str,
"time": time_str,
"week_num": str(weekday_index + 1),
"week_cn": weekdays_cn[weekday_index],
"week_short_cn": weekdays_short[weekday_index],
"week_en": weekdays_en[weekday_index]
}
return success_response(formatted_data, "Time retrieved successfully")
except Exception as e:
print(f"Error parsing time response: {e}")
return success_response(get_local_time(), "Time retrieved from local system (fallback)")
print("Warning: Time API unavailable, using local time")
return success_response(get_local_time(), "Time retrieved from local system")
@app.route('/api/v3/weather/sojson', methods=['GET'])
def get_weather_sojson():
"""
Get weather (sojson API, uses citykey)
GET /api/v3/weather/sojson?city=Linyi
"""
city_name = request.args.get('city', '').strip()
if not city_name:
return error_response(400, "Please provide city name parameter ?city=city_name")
city_code, error_msg = search_city_code(city_name)
if error_msg:
return error_response(404, error_msg)
url = WEATHER_SOJSON_URL.format(city_code)
result = safe_request(url)
if result and result.get("status") == 200:
return success_response(result, "Weather retrieved successfully")
elif result and "error" in result:
return error_response(500, result.get("message", "Weather service error"), 500)
else:
return error_response(500, "Weather service temporarily unavailable", 500)
@app.route('/api/v3/weather', methods=['GET'])
def get_weather_uapis():
"""
Get weather (uapis.cn general weather API)
GET /api/v3/weather?city=Beijing
GET /api/v3/weather?adcode=110000
GET /api/v3/weather (auto IP geolocation)
"""
params = {}
for key in ['city', 'adcode', 'extended', 'forecast', 'hourly', 'minutely', 'indices', 'lang']:
value = request.args.get(key)
if value is not None:
if key in ['extended', 'forecast', 'hourly', 'minutely', 'indices']:
params[key] = value.lower() == 'true'
else:
params[key] = value
result = safe_request(WEATHER_UAPIS_URL, params=params)
if result:
if "error" in result:
return error_response(500, result.get("message", "Weather service error"), 500)
return success_response(result, "Weather retrieved successfully")
else:
return error_response(500, "Weather service temporarily unavailable", 500)
@app.route('/api/v3/phone', methods=['GET'])
def get_phone_info():
"""
Query phone number location
GET /api/v3/phone?phone=13800138000
"""
phone = request.args.get('phone', '').strip()
if not phone:
return error_response(400, "Please provide phone number parameter ?phone=phone_number")
if not re.match(r'^1[3-9]\d{9}$', phone):
return error_response(400, "Invalid phone number format, please enter an 11-digit Chinese mainland phone number")
result = safe_request(PHONE_INFO_URL, params={"phone": phone})
if result:
if "error" in result:
return error_response(400, result.get("message", "Query failed"), 400)
return success_response(result, "Phone number lookup successful")
else:
return error_response(500, "Phone number lookup service temporarily unavailable", 500)
# ========== IP Geolocation Routes ==========
@app.route('/api/v3/ip/me', methods=['GET'])
def get_my_ip():
"""
Get current client's public IP address
GET /api/v3/ip/me
GET /api/v3/ip/me?source=commercial (returns more detailed info)
"""
source = request.args.get('source', '')
params = {}
if source:
params['source'] = source
result = safe_request(IP_MYIP_URL, params=params if params else None)
if result and "ip" in result:
# Format the response
formatted_data = {
"ip": result.get("ip", ""),
"region": result.get("region", ""),
"isp": result.get("isp", ""),
"llc": result.get("llc", ""),
"asn": result.get("asn", ""),
"latitude": result.get("latitude", ""),
"longitude": result.get("longitude", ""),
"beginip": result.get("beginip", ""),
"endip": result.get("endip", "")
}
# Add commercial fields if available
if "district" in result:
formatted_data["district"] = result.get("district", "")
if "time_zone" in result:
formatted_data["time_zone"] = result.get("time_zone", "")
return success_response(formatted_data, "IP geolocation successful")
elif result and "error" in result:
return error_response(500, result.get("message", "IP service error"), 500)
else:
return error_response(500, "IP geolocation service temporarily unavailable", 500)
@app.route('/api/v3/ip', methods=['GET'])
def get_ip_info():
"""
Query IP information
GET /api/v3/ip?ip=8.8.8.8
GET /api/v3/ip?ip=8.8.8.8&source=commercial
GET /api/v3/ip (auto-detect current IP)
"""
ip = request.args.get('ip', '').strip()
source = request.args.get('source', '')
# If no IP provided, use myip endpoint
if not ip:
# Use IP_SB as fallback for auto-detection
try:
resp = requests.get("https://api.ip.sb/geoip", timeout=5)
if resp.status_code == 200:
data = resp.json()
formatted_data = {
"ip": data.get("ip", ""),
"region": f"{data.get('country', '')} {data.get('region', '')} {data.get('city', '')}".strip(),
"isp": data.get("isp", ""),
"organization": data.get("organization", ""),
"asn": data.get("asn", ""),
"latitude": data.get("latitude", ""),
"longitude": data.get("longitude", ""),
"country": data.get("country", ""),
"country_code": data.get("country_code", ""),
"region_code": data.get("region_code", ""),
"city": data.get("city", "")
}
return success_response(formatted_data, "IP geolocation successful")
except:
pass
# Fallback to uapis myip
result = safe_request(IP_MYIP_URL)
if result and "ip" in result:
formatted_data = {
"ip": result.get("ip", ""),
"region": result.get("region", ""),
"isp": result.get("isp", ""),
"llc": result.get("llc", ""),
"asn": result.get("asn", ""),
"latitude": result.get("latitude", ""),
"longitude": result.get("longitude", ""),
"beginip": result.get("beginip", ""),
"endip": result.get("endip", "")
}
return success_response(formatted_data, "IP geolocation successful")
return error_response(500, "Unable to determine IP address", 500)
# Query specific IP
# Try uapis first
params = {"ip": ip}
if source:
params["source"] = source
result = safe_request(IP_INFO_URL, params=params)
if result and "ip" in result:
formatted_data = {
"ip": result.get("ip", ""),
"region": result.get("region", ""),
"isp": result.get("isp", ""),
"llc": result.get("llc", ""),
"asn": result.get("asn", ""),
"latitude": result.get("latitude", ""),
"longitude": result.get("longitude", ""),
"beginip": result.get("beginip", ""),
"endip": result.get("endip", "")
}
return success_response(formatted_data, "IP geolocation successful")
# Fallback to ip.sb
try:
url = IP_SB_URL.format(ip)
resp = requests.get(url, timeout=5)
if resp.status_code == 200:
data = resp.json()
formatted_data = {
"ip": data.get("ip", ip),
"region": f"{data.get('country', '')} {data.get('region', '')} {data.get('city', '')}".strip(),
"isp": data.get("isp", ""),
"organization": data.get("organization", ""),
"asn": data.get("asn", ""),
"latitude": data.get("latitude", ""),
"longitude": data.get("longitude", ""),
"country": data.get("country", ""),
"country_code": data.get("country_code", ""),
"region_code": data.get("region_code", ""),
"city": data.get("city", "")
}
return success_response(formatted_data, "IP geolocation successful")
except:
pass
return error_response(500, "IP geolocation service temporarily unavailable", 500)
@app.route('/api/v3/ip/advanced', methods=['GET'])
def get_ip_advanced():
"""
Advanced IP query with commercial data source
GET /api/v3/ip/advanced?ip=8.8.8.8
"""
ip = request.args.get('ip', '').strip()
source = request.args.get('source', 'commercial')
if not ip:
return error_response(400, "Please provide IP address parameter ?ip=8.8.8.8")
params = {"ip": ip, "source": source}
result = safe_request(IP_INFO_URL, params=params)
if result and "ip" in result:
return success_response(result, "IP geolocation successful")
elif result and "error" in result:
return error_response(500, result.get("message", "IP service error"), 500)
else:
return error_response(500, "IP geolocation service temporarily unavailable", 500)
@app.route('/api/v3', methods=['GET'])
def api_index():
"""API directory"""
return jsonify({
"code": 200,
"message": "Unified API Gateway v3.0",
"version": "3.0.0",
"endpoints": {
"time": {
"path": "/api/v3/time",
"method": "GET",
"description": "Get current time",
"example": "/api/v3/time"
},
"weather_sojson": {
"path": "/api/v3/weather/sojson",
"method": "GET",
"description": "Get weather (sojson, uses citykey)",
"params": {"city": "City name"},
"example": "/api/v3/weather/sojson?city=Linyi"
},
"weather": {
"path": "/api/v3/weather",
"method": "GET",
"description": "Get weather (uapis.cn general API)",
"params": {"city": "City name", "forecast": "Enable forecast", "extended": "Enable extended fields"},
"example": "/api/v3/weather?city=Beijing&forecast=true"
},
"phone": {
"path": "/api/v3/phone",
"method": "GET",
"description": "Query phone number location",
"params": {"phone": "Phone number"},
"example": "/api/v3/phone?phone=13800138000"
},
"ip_me": {
"path": "/api/v3/ip/me",
"method": "GET",
"description": "Get your public IP address",
"params": {"source": "commercial (optional)"},
"example": "/api/v3/ip/me"
},
"ip": {
"path": "/api/v3/ip",
"method": "GET",
"description": "Query IP information",
"params": {"ip": "IP address (optional)", "source": "commercial (optional)"},
"example": "/api/v3/ip?ip=8.8.8.8"
},
"ip_advanced": {
"path": "/api/v3/ip/advanced",
"method": "GET",
"description": "Advanced IP query with commercial data",
"params": {"ip": "IP address", "source": "commercial (default)"},
"example": "/api/v3/ip/advanced?ip=8.8.8.8"
}
}
})
# ========== Error Handlers ==========
@app.errorhandler(404)
def not_found(error):
return jsonify({
"code": 404,
"message": "Endpoint not found, visit /api/v3 to see available endpoints"
}), 404
@app.errorhandler(500)
def server_error(error):
return jsonify({
"code": 500,
"message": "Internal server error"
}), 500
# ========== Startup ==========
if __name__ == '__main__':
load_city_codes()
app.run(host='0.0.0.0', port=880, debug=False)