增加西部数码域名注册api接口文件及端点

This commit is contained in:
jingrow 2025-07-31 17:07:49 +08:00
parent 88df4fbbf6
commit 586a4effcf
3 changed files with 607 additions and 16 deletions

560
jcloud/api/domain_west.py Normal file
View File

@ -0,0 +1,560 @@
# Copyright (c) 2024, JINGROW
# For license information, please see license.txt
import jingrow
import requests
import time
import hashlib
import json
from urllib.parse import urlencode
from typing import Dict, Any, Optional, List
class WestDomainAPI:
"""西部数码域名API客户端"""
def __init__(self, username: str, password: str):
"""
初始化西部数码API客户端
Args:
username: 西部数码用户名
password: 西部数码API密码
"""
self.username = username.strip()
self.password = password.strip()
self.api_base_url = "https://api.west.cn/api/v2"
self.time = None
self.token = None
def _generate_token(self) -> str:
"""生成认证token"""
self.time = self._get_current_timestamp()
token_string = f"{self.username}{self.password}{self.time}"
return hashlib.md5(token_string.encode('utf-8')).hexdigest()
def _get_current_timestamp(self) -> int:
"""获取当前时间戳(毫秒)"""
return int(time.time() * 1000)
def _generate_common_parameters(self) -> Dict[str, str]:
"""生成公共参数"""
self.token = self._generate_token()
return {
'username': self.username,
'time': str(self.time),
'token': self.token,
}
def _make_request(self, action: str, method: str = 'GET',
query_params: Optional[Dict] = None,
body_params: Optional[Dict] = None) -> Dict[str, Any]:
"""
发送API请求
Args:
action: API动作路径
method: 请求方法 (GET/POST)
query_params: 查询参数
body_params: 请求体参数
Returns:
API响应结果
"""
# 构建URL
common_params = self._generate_common_parameters()
param_string = urlencode(common_params)
url = f"{self.api_base_url}{action}"
if '?' in action:
url += f"&{param_string}"
else:
url += f"?{param_string}"
# 添加查询参数
if query_params:
url += f"&{urlencode(query_params)}"
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
try:
if method.upper() == 'POST':
data = body_params or {}
response = requests.post(url, data=data, headers=headers, timeout=30)
else:
response = requests.get(url, headers=headers, timeout=30)
response.raise_for_status()
try:
result = response.json()
except json.JSONDecodeError:
jingrow.log_error("西部数码API响应解析失败", response_text=response.text)
result = {"status": "error", "message": "无法解析API响应"}
return result
except requests.exceptions.RequestException as e:
jingrow.log_error("西部数码API请求失败", error=str(e), url=url)
return {"status": "error", "message": f"API请求失败: {str(e)}"}
def check_balance(self) -> Dict[str, Any]:
"""获取账户可用余额"""
return self._make_request('/info/?act=checkbalance', 'GET')
def get_domain_price(self, domain: str, year: int = 1) -> Dict[str, Any]:
"""
获取域名价格
Args:
domain: 域名
year: 注册年限
"""
body_params = {
'type': 'domain',
'value': domain,
'year': year,
}
return self._make_request('/info/?act=getprice', 'POST', body_params=body_params)
def query_domain(self, domain: str, suffix: str = '.com') -> Dict[str, Any]:
"""
域名查询
Args:
domain: 域名前缀
suffix: 域名后缀
"""
body_params = {
'domain': domain,
'suffix': suffix,
}
return self._make_request('/domain/query/', 'POST', body_params=body_params)
def register_domain(self, domain: str, year: int = 1,
contact_info: Optional[Dict] = None) -> Dict[str, Any]:
"""
注册域名
Args:
domain: 域名
year: 注册年限
contact_info: 联系人信息
"""
body_params = {
'domain': domain,
'year': year,
}
if contact_info:
body_params.update(contact_info)
return self._make_request('/domain/?act=register', 'POST', body_params=body_params)
def renew_domain(self, domain: str, year: int = 1,
expire_date: Optional[str] = None,
client_price: Optional[int] = None) -> Dict[str, Any]:
"""
域名续费
Args:
domain: 域名
year: 续费年限
expire_date: 到期时间
client_price: 客户价格
"""
body_params = {
'domain': domain,
'year': year,
}
if expire_date:
body_params['expiredate'] = expire_date
if client_price:
body_params['client_price'] = client_price
return self._make_request('/domain/?act=renew', 'POST', body_params=body_params)
def get_domain_list(self, limit: int = 10, page: int = 1) -> Dict[str, Any]:
"""
获取域名列表
Args:
limit: 每页数量
page: 页码
"""
query_params = {
'limit': limit,
'page': page,
}
return self._make_request('/domain/?act=getdomains', 'GET', query_params=query_params)
def get_domain_info(self, domain: str) -> Dict[str, Any]:
"""
获取域名详细信息
Args:
domain: 域名
"""
body_params = {
'domain': domain,
}
return self._make_request('/domain/?act=getinfo', 'POST', body_params=body_params)
def get_dns_records(self, domain: str) -> Dict[str, Any]:
"""
获取域名DNS记录
Args:
domain: 域名
"""
body_params = {
'domain': domain,
}
return self._make_request('/domain/?act=getdns', 'POST', body_params=body_params)
def modify_dns_records(self, domain: str, records: List[Dict]) -> Dict[str, Any]:
"""
修改域名DNS记录
Args:
domain: 域名
records: DNS记录列表
"""
body_params = {
'domain': domain,
'records': json.dumps(records),
}
return self._make_request('/domain/?act=modifydns', 'POST', body_params=body_params)
def add_dns_record(self, domain: str, record_type: str,
host: str, value: str, ttl: int = 600) -> Dict[str, Any]:
"""
添加DNS记录
Args:
domain: 域名
record_type: 记录类型 (A, CNAME, MX, TXT等)
host: 主机记录
value: 记录值
ttl: TTL值
"""
record = {
'type': record_type,
'host': host,
'value': value,
'ttl': ttl,
}
return self.modify_dns_records(domain, [record])
def delete_dns_record(self, domain: str, record_id: str) -> Dict[str, Any]:
"""
删除DNS记录
Args:
domain: 域名
record_id: 记录ID
"""
body_params = {
'domain': domain,
'record_id': record_id,
}
return self._make_request('/domain/?act=deletedns', 'POST', body_params=body_params)
def transfer_domain(self, domain: str, auth_code: str) -> Dict[str, Any]:
"""
域名转入
Args:
domain: 域名
auth_code: 转移授权码
"""
body_params = {
'domain': domain,
'auth_code': auth_code,
}
return self._make_request('/domain/?act=transfer', 'POST', body_params=body_params)
def lock_domain(self, domain: str, lock: bool = True) -> Dict[str, Any]:
"""
锁定/解锁域名
Args:
domain: 域名
lock: 是否锁定
"""
action = '/domain/?act=lock' if lock else '/domain/?act=unlock'
body_params = {
'domain': domain,
}
return self._make_request(action, 'POST', body_params=body_params)
def get_west_domain_client() -> WestDomainAPI:
"""获取西部数码域名API客户端实例"""
try:
# 从Jcloud Settings获取配置
settings = jingrow.get_single("Jcloud Settings")
if settings:
username = settings.get("west_username")
password = settings.get_password("west_api_password") if settings.get("west_api_password") else None
if username and password:
return WestDomainAPI(username, password)
else:
jingrow.log_error("西部数码域名API: 缺少用户名或密码配置")
return None
else:
jingrow.log_error("西部数码域名API: Jcloud Settings 尚未配置")
return None
except Exception as e:
jingrow.log_error(f"西部数码域名API客户端初始化失败: {str(e)}")
return None
# API端点函数
@jingrow.whitelist()
def check_west_balance():
"""检查西部数码账户余额"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
return client.check_balance()
@jingrow.whitelist()
def check_domain(**data):
"""查询域名是否可注册"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
suffix = data.get('suffix', '.com')
if not domain:
return {"status": "error", "message": "缺少域名参数"}
return client.query_domain(domain, suffix)
@jingrow.whitelist()
def west_domain_get_price(**data):
"""获取域名价格"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
year = data.get('year', 1)
if not domain:
return {"status": "error", "message": "缺少域名参数"}
return client.get_domain_price(domain, year)
@jingrow.whitelist()
def west_domain_register(**data):
"""注册域名"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
year = data.get('year', 1)
contact_info = data.get('contact_info')
if not domain:
return {"status": "error", "message": "缺少域名参数"}
return client.register_domain(domain, year, contact_info)
@jingrow.whitelist()
def west_domain_renew(**data):
"""域名续费"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
year = data.get('year', 1)
expire_date = data.get('expire_date')
client_price = data.get('client_price')
if not domain:
return {"status": "error", "message": "缺少域名参数"}
return client.renew_domain(domain, year, expire_date, client_price)
@jingrow.whitelist()
def west_domain_get_list(**data):
"""获取域名列表"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
limit = data.get('limit', 10)
page = data.get('page', 1)
return client.get_domain_list(limit, page)
@jingrow.whitelist()
def west_domain_get_info(**data):
"""获取域名信息"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
if not domain:
return {"status": "error", "message": "缺少域名参数"}
return client.get_domain_info(domain)
@jingrow.whitelist()
def west_domain_get_dns(**data):
"""获取域名DNS记录"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
if not domain:
return {"status": "error", "message": "缺少域名参数"}
return client.get_dns_records(domain)
@jingrow.whitelist()
def west_domain_modify_dns(**data):
"""修改域名DNS记录"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
records = data.get('records')
if not domain:
return {"status": "error", "message": "缺少域名参数"}
if not records:
return {"status": "error", "message": "缺少DNS记录参数"}
return client.modify_dns_records(domain, records)
@jingrow.whitelist()
def west_domain_add_dns_record(**data):
"""添加DNS记录"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
record_type = data.get('record_type')
host = data.get('host')
value = data.get('value')
ttl = data.get('ttl', 600)
if not all([domain, record_type, host, value]):
return {"status": "error", "message": "缺少必要参数"}
return client.add_dns_record(domain, record_type, host, value, ttl)
@jingrow.whitelist()
def west_domain_delete_dns_record(**data):
"""删除DNS记录"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
record_id = data.get('record_id')
if not domain:
return {"status": "error", "message": "缺少域名参数"}
if not record_id:
return {"status": "error", "message": "缺少记录ID参数"}
return client.delete_dns_record(domain, record_id)
@jingrow.whitelist()
def west_domain_transfer(**data):
"""域名转入"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
auth_code = data.get('auth_code')
if not domain:
return {"status": "error", "message": "缺少域名参数"}
if not auth_code:
return {"status": "error", "message": "缺少转移授权码参数"}
return client.transfer_domain(domain, auth_code)
@jingrow.whitelist()
def west_domain_lock(**data):
"""锁定域名"""
client = get_west_domain_client()
if not client:
return {"status": "error", "message": "API客户端初始化失败"}
domain = data.get('domain')
lock = data.get('lock', True)
if not domain:
return {"status": "error", "message": "缺少域名参数"}
return client.lock_domain(domain, lock)
# 便捷函数
def call_west_domain_api(api_name: str, **kwargs) -> Dict[str, Any]:
"""
调用西部数码域名API的通用函数
Args:
api_name: API名称
**kwargs: API参数
Returns:
API响应结果
"""
api_functions = {
'check_balance': check_west_balance,
'query': check_domain,
'get_price': west_domain_get_price,
'register': west_domain_register,
'renew': west_domain_renew,
'get_list': west_domain_get_list,
'get_info': west_domain_get_info,
'get_dns': west_domain_get_dns,
'modify_dns': west_domain_modify_dns,
'add_dns_record': west_domain_add_dns_record,
'delete_dns_record': west_domain_delete_dns_record,
'transfer': west_domain_transfer,
'lock': west_domain_lock,
}
if api_name not in api_functions:
return {"status": "error", "message": f"未知的API: {api_name}"}
try:
return api_functions[api_name](**kwargs)
except Exception as e:
jingrow.log_error(f"西部数码域名API调用失败: {api_name}", error=str(e), params=kwargs)
return {"status": "error", "message": f"API调用失败: {str(e)}"}

View File

@ -1,7 +1,6 @@
{
"actions": [],
"creation": "2022-02-08 15:13:48.372783",
"pagetype": "PageType",
"editable_grid": 1,
"engine": "InnoDB",
"field_order": [
@ -147,7 +146,12 @@
"telegram_bot_token",
"section_break_vvyh",
"aliyun_access_key_id",
"column_break_ssjr",
"aliyun_access_secret",
"west_section",
"west_username",
"column_break_knph",
"west_api_password",
"mailgun_settings_section",
"mailgun_api_key",
"root_domain",
@ -1420,25 +1424,25 @@
},
{
"default": "https://openapi.alipay.com/gateway.do",
"description": "\u652f\u4ed8\u5b9d\u7f51\u5173\u5730\u5740\uff0c\u4f8b\u5982\uff1ahttps://openapi.alipay.com/gateway.do",
"description": "支付宝网关地址,例如:https://openapi.alipay.com/gateway.do",
"fieldname": "alipay_server_url",
"fieldtype": "Data",
"label": "Alipay Server URL"
},
{
"description": "\u652f\u4ed8\u5b9d\u5e94\u7528ID",
"description": "支付宝应用ID",
"fieldname": "alipay_app_id",
"fieldtype": "Data",
"label": "Alipay App ID"
},
{
"description": "\u652f\u4ed8\u5b8c\u6210\u540e\u7684\u8df3\u8f6c\u5730\u5740",
"description": "支付完成后的跳转地址",
"fieldname": "alipay_return_url",
"fieldtype": "Data",
"label": "Alipay Return URL"
},
{
"description": "\u652f\u4ed8\u7ed3\u679c\u901a\u77e5\u5730\u5740",
"description": "支付结果通知地址",
"fieldname": "alipay_notify_url",
"fieldtype": "Data",
"label": "Alipay Notify URL"
@ -1448,13 +1452,13 @@
"fieldtype": "Column Break"
},
{
"description": "\u5e94\u7528\u79c1\u94a5\uff0c\u7528\u4e8e\u7b7e\u540d",
"description": "应用私钥,用于签名",
"fieldname": "alipay_app_private_key",
"fieldtype": "Long Text",
"label": "Alipay App Private Key"
},
{
"description": "\u652f\u4ed8\u5b9d\u516c\u94a5\uff0c\u7528\u4e8e\u9a8c\u8bc1\u7b7e\u540d",
"description": "支付宝公钥,用于验证签名",
"fieldname": "alipay_public_key",
"fieldtype": "Long Text",
"label": "Alipay Public Key"
@ -1466,25 +1470,25 @@
"label": "Wechatpay Settings"
},
{
"description": "\u5fae\u4fe1\u652f\u4ed8AppID",
"description": "微信支付AppID",
"fieldname": "wechatpay_appid",
"fieldtype": "Data",
"label": "Wechatpay App ID"
},
{
"description": "\u5fae\u4fe1\u652f\u4ed8\u5546\u6237\u53f7",
"description": "微信支付商户号",
"fieldname": "wechatpay_mchid",
"fieldtype": "Data",
"label": "Wechatpay Merchant ID"
},
{
"description": "\u5fae\u4fe1\u652f\u4ed8\u7ed3\u679c\u901a\u77e5\u5730\u5740",
"description": "微信支付结果通知地址",
"fieldname": "wechatpay_notify_url",
"fieldtype": "Data",
"label": "Wechatpay Notify URL"
},
{
"description": "\u8bc1\u4e66\u5e8f\u5217\u53f7",
"description": "证书序列号",
"fieldname": "wechatpay_cert_serial_no",
"fieldtype": "Data",
"label": "Wechatpay Certificate Serial Number"
@ -1494,25 +1498,25 @@
"fieldtype": "Column Break"
},
{
"description": "\u5fae\u4fe1\u652f\u4ed8APIv3\u5bc6\u94a5",
"description": "微信支付APIv3密钥",
"fieldname": "wechatpay_apiv3_key",
"fieldtype": "Data",
"label": "Wechatpay API v3 Key"
},
{
"description": "\u5fae\u4fe1\u652f\u4ed8\u5546\u6237\u79c1\u94a5",
"description": "微信支付商户私钥",
"fieldname": "wechatpay_private_key",
"fieldtype": "Long Text",
"label": "Wechatpay Private Key"
},
{
"description": "\u5fae\u4fe1\u652f\u4ed8\u5e73\u53f0\u516c\u94a5",
"description": "微信支付平台公钥",
"fieldname": "wechatpay_public_key",
"fieldtype": "Long Text",
"label": "Wechatpay Public Key"
},
{
"description": "\u5fae\u4fe1\u652f\u4ed8\u516c\u94a5ID",
"description": "微信支付公钥ID",
"fieldname": "wechatpay_public_key_id",
"fieldtype": "Data",
"label": "Wechatpay Public Key ID"
@ -1570,15 +1574,39 @@
{
"fieldname": "column_break_jhbn",
"fieldtype": "Column Break"
},
{
"fieldname": "column_break_ssjr",
"fieldtype": "Column Break"
},
{
"fieldname": "west_section",
"fieldtype": "Section Break",
"label": "West"
},
{
"fieldname": "column_break_knph",
"fieldtype": "Column Break"
},
{
"fieldname": "west_username",
"fieldtype": "Data",
"label": "West Username"
},
{
"fieldname": "west_api_password",
"fieldtype": "Password",
"label": "West Api Password"
}
],
"issingle": 1,
"links": [],
"modified": "2025-04-06 19:58:13.368427",
"modified": "2025-07-31 17:02:23.928563",
"modified_by": "Administrator",
"module": "Jcloud",
"name": "Jcloud Settings",
"owner": "Administrator",
"pagetype": "PageType",
"permissions": [
{
"create": 1,
@ -1592,6 +1620,7 @@
}
],
"quick_entry": 1,
"row_format": "Dynamic",
"sort_field": "modified",
"sort_order": "DESC",
"states": [],

View File

@ -184,6 +184,8 @@ class JcloudSettings(Document):
wechatpay_private_key: DF.LongText | None
wechatpay_public_key: DF.LongText | None
wechatpay_public_key_id: DF.Data | None
west_api_password: DF.Password | None
west_username: DF.Data | None
# end: auto-generated types
dashboard_fields = (