e171440ffd
- Add Dockerfile with Python 3.13 + uv - Add Gitea Actions workflow for auto-build on push - Add deposit_api.py for balance management - Update api_server.py with domain registration endpoint Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
524 lines
16 KiB
Python
524 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Namecheap REST API Server for OpenAI Function Calling
|
|
"""
|
|
|
|
import os
|
|
from typing import Optional
|
|
from dotenv import load_dotenv
|
|
from fastapi import FastAPI, HTTPException, Header, Depends
|
|
from pydantic import BaseModel
|
|
|
|
from namecheap import NamecheapAPI, NamecheapConfig, RegistrantInfo
|
|
|
|
load_dotenv()
|
|
|
|
# API Key for authentication
|
|
API_KEY = os.getenv("API_SERVER_KEY", "")
|
|
|
|
|
|
def verify_api_key(x_api_key: str = Header(None)):
|
|
if API_KEY and x_api_key != API_KEY:
|
|
raise HTTPException(status_code=401, detail="Invalid API key")
|
|
|
|
|
|
app = FastAPI(
|
|
title="Namecheap API",
|
|
description="REST API for Namecheap domain management",
|
|
version="1.0.0",
|
|
dependencies=[Depends(verify_api_key)],
|
|
)
|
|
|
|
# Initialize Namecheap API
|
|
config = NamecheapConfig(
|
|
api_user=os.getenv("NAMECHEAP_API_USER", ""),
|
|
api_key=os.getenv("NAMECHEAP_API_KEY", ""),
|
|
username=os.getenv("NAMECHEAP_USERNAME", ""),
|
|
client_ip=os.getenv("NAMECHEAP_CLIENT_IP", ""),
|
|
sandbox=os.getenv("NAMECHEAP_SANDBOX", "false").lower() == "true",
|
|
)
|
|
api = NamecheapAPI(config)
|
|
|
|
|
|
# ===== Models =====
|
|
|
|
class DomainCheckRequest(BaseModel):
|
|
domains: list[str]
|
|
|
|
class DomainRenewRequest(BaseModel):
|
|
domain: str
|
|
years: int = 1
|
|
|
|
class NameserverRequest(BaseModel):
|
|
domain: str
|
|
nameservers: list[str]
|
|
|
|
|
|
class GlueRecordRequest(BaseModel):
|
|
nameserver: str
|
|
ip: str
|
|
|
|
|
|
class GlueRecordUpdateRequest(BaseModel):
|
|
nameserver: str
|
|
old_ip: str
|
|
ip: str
|
|
|
|
class ContactRequest(BaseModel):
|
|
domain: str
|
|
organization: str = ""
|
|
first_name: str
|
|
last_name: str
|
|
address1: str
|
|
address2: str = ""
|
|
city: str
|
|
state_province: str
|
|
postal_code: str
|
|
country: str
|
|
phone: str
|
|
email: str
|
|
|
|
|
|
# ===== Domain Endpoints =====
|
|
|
|
@app.get("/domains")
|
|
def list_domains(page: int = 1, page_size: int = 20):
|
|
"""Get list of domains in account"""
|
|
return api.domains_get_list(page=page, page_size=page_size)
|
|
|
|
|
|
@app.post("/domains/check")
|
|
def check_domains(req: DomainCheckRequest):
|
|
"""Check domain availability"""
|
|
return api.domains_check(req.domains)
|
|
|
|
|
|
@app.get("/domains/{domain}")
|
|
def get_domain_info(domain: str):
|
|
"""Get detailed info about a domain"""
|
|
return api.domains_get_info(domain)
|
|
|
|
|
|
@app.post("/domains/{domain}/renew")
|
|
def renew_domain(domain: str, years: int = 1):
|
|
"""Renew a domain. WARNING: This will charge your account."""
|
|
return api.domains_renew(domain, years=years)
|
|
|
|
|
|
@app.get("/domains/{domain}/contacts")
|
|
def get_domain_contacts(domain: str):
|
|
"""Get domain contact information"""
|
|
return api.domains_get_contacts(domain)
|
|
|
|
|
|
@app.put("/domains/{domain}/contacts")
|
|
def set_domain_contacts(domain: str, req: ContactRequest):
|
|
"""Set domain contact information"""
|
|
registrant = RegistrantInfo(
|
|
organization=req.organization,
|
|
first_name=req.first_name,
|
|
last_name=req.last_name,
|
|
address1=req.address1,
|
|
address2=req.address2,
|
|
city=req.city,
|
|
state_province=req.state_province,
|
|
postal_code=req.postal_code,
|
|
country=req.country,
|
|
phone=req.phone,
|
|
email=req.email,
|
|
)
|
|
success = api.domains_set_contacts(domain, registrant)
|
|
return {"success": success}
|
|
|
|
|
|
# ===== DNS Endpoints =====
|
|
|
|
@app.get("/dns/{domain}/nameservers")
|
|
def get_nameservers(domain: str):
|
|
"""Get nameserver information for a domain"""
|
|
parts = domain.rsplit(".", 1)
|
|
if len(parts) != 2:
|
|
raise HTTPException(status_code=400, detail="Invalid domain format")
|
|
sld, tld = parts
|
|
return api.dns_get_list(sld, tld)
|
|
|
|
|
|
@app.put("/dns/{domain}/nameservers")
|
|
def set_nameservers(domain: str, req: NameserverRequest):
|
|
"""Set custom nameservers for a domain"""
|
|
from namecheap import NamecheapError
|
|
parts = domain.rsplit(".", 1)
|
|
if len(parts) != 2:
|
|
raise HTTPException(status_code=400, detail="Invalid domain format")
|
|
sld, tld = parts
|
|
try:
|
|
success = api.dns_set_custom(sld, tld, req.nameservers)
|
|
return {"success": success, "nameservers": req.nameservers}
|
|
except NamecheapError as e:
|
|
error_msg = str(e)
|
|
if "subordinate" in error_msg.lower() or "non existen" in error_msg.lower():
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"네임서버 {req.nameservers}는 등록되지 않았습니다. 자기 도메인을 네임서버로 사용하려면 먼저 Child Nameserver(글루 레코드)를 IP 주소와 함께 등록해야 합니다."
|
|
)
|
|
raise HTTPException(status_code=400, detail=error_msg)
|
|
|
|
|
|
@app.post("/dns/{domain}/nameservers/default")
|
|
def set_default_nameservers(domain: str):
|
|
"""Set default Namecheap nameservers"""
|
|
parts = domain.rsplit(".", 1)
|
|
if len(parts) != 2:
|
|
raise HTTPException(status_code=400, detail="Invalid domain format")
|
|
sld, tld = parts
|
|
success = api.dns_set_default(sld, tld)
|
|
return {"success": success}
|
|
|
|
|
|
@app.get("/dns/{domain}/records")
|
|
def get_dns_records(domain: str):
|
|
"""Get DNS records. Only works if using Namecheap DNS."""
|
|
parts = domain.rsplit(".", 1)
|
|
if len(parts) != 2:
|
|
raise HTTPException(status_code=400, detail="Invalid domain format")
|
|
sld, tld = parts
|
|
return api.dns_get_hosts(sld, tld)
|
|
|
|
|
|
# ===== Glue Record (Child Nameserver) Endpoints =====
|
|
|
|
@app.post("/dns/{domain}/glue")
|
|
def create_glue_record(domain: str, req: GlueRecordRequest):
|
|
"""Create a glue record (child nameserver) for a domain"""
|
|
from namecheap import NamecheapError
|
|
parts = domain.rsplit(".", 1)
|
|
if len(parts) != 2:
|
|
raise HTTPException(status_code=400, detail="Invalid domain format")
|
|
sld, tld = parts
|
|
try:
|
|
return api.ns_create(sld, tld, req.nameserver, req.ip)
|
|
except NamecheapError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@app.get("/dns/{domain}/glue/{nameserver}")
|
|
def get_glue_record(domain: str, nameserver: str):
|
|
"""Get info about a glue record (child nameserver)"""
|
|
from namecheap import NamecheapError
|
|
parts = domain.rsplit(".", 1)
|
|
if len(parts) != 2:
|
|
raise HTTPException(status_code=400, detail="Invalid domain format")
|
|
sld, tld = parts
|
|
try:
|
|
return api.ns_get_info(sld, tld, nameserver)
|
|
except NamecheapError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@app.put("/dns/{domain}/glue")
|
|
def update_glue_record(domain: str, req: GlueRecordUpdateRequest):
|
|
"""Update a glue record (child nameserver) IP address"""
|
|
from namecheap import NamecheapError
|
|
parts = domain.rsplit(".", 1)
|
|
if len(parts) != 2:
|
|
raise HTTPException(status_code=400, detail="Invalid domain format")
|
|
sld, tld = parts
|
|
try:
|
|
return api.ns_update(sld, tld, req.nameserver, req.old_ip, req.ip)
|
|
except NamecheapError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@app.delete("/dns/{domain}/glue/{nameserver}")
|
|
def delete_glue_record(domain: str, nameserver: str):
|
|
"""Delete a glue record (child nameserver)"""
|
|
from namecheap import NamecheapError
|
|
parts = domain.rsplit(".", 1)
|
|
if len(parts) != 2:
|
|
raise HTTPException(status_code=400, detail="Invalid domain format")
|
|
sld, tld = parts
|
|
try:
|
|
return api.ns_delete(sld, tld, nameserver)
|
|
except NamecheapError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
# ===== Account Endpoints =====
|
|
|
|
@app.get("/account/balance")
|
|
def get_balance():
|
|
"""Get account balance"""
|
|
return api.users_get_balances()
|
|
|
|
|
|
@app.get("/prices/{tld}")
|
|
def get_price(tld: str):
|
|
"""Get registration price for a TLD"""
|
|
from db import get_prices, init_db
|
|
init_db()
|
|
|
|
prices = get_prices()
|
|
for p in prices:
|
|
if p["tld"] == tld:
|
|
return {"tld": tld, "usd": p["usd"], "krw": p["krw"]}
|
|
|
|
raise HTTPException(status_code=404, detail=f"TLD '{tld}' not found")
|
|
|
|
|
|
@app.get("/prices")
|
|
def list_prices(limit: int = 50):
|
|
"""Get all TLD prices"""
|
|
from db import get_prices, init_db
|
|
init_db()
|
|
return get_prices()[:limit]
|
|
|
|
|
|
# ===== OpenAI Schema =====
|
|
|
|
@app.get("/openai/schema")
|
|
def get_openai_schema():
|
|
"""Get OpenAI Function Calling schema"""
|
|
return {
|
|
"functions": [
|
|
{
|
|
"name": "list_domains",
|
|
"description": "Get list of domains in account",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"page": {"type": "integer", "default": 1},
|
|
"page_size": {"type": "integer", "default": 20}
|
|
}
|
|
}
|
|
},
|
|
{
|
|
"name": "check_domains",
|
|
"description": "Check domain availability",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"domains": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"description": "List of domains to check"
|
|
}
|
|
},
|
|
"required": ["domains"]
|
|
}
|
|
},
|
|
{
|
|
"name": "get_domain_info",
|
|
"description": "Get detailed info about a domain",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"domain": {"type": "string"}
|
|
},
|
|
"required": ["domain"]
|
|
}
|
|
},
|
|
{
|
|
"name": "get_nameservers",
|
|
"description": "Get nameserver information for a domain",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"domain": {"type": "string"}
|
|
},
|
|
"required": ["domain"]
|
|
}
|
|
},
|
|
{
|
|
"name": "set_nameservers",
|
|
"description": "Set custom nameservers for a domain",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"domain": {"type": "string"},
|
|
"nameservers": {
|
|
"type": "array",
|
|
"items": {"type": "string"}
|
|
}
|
|
},
|
|
"required": ["domain", "nameservers"]
|
|
}
|
|
},
|
|
{
|
|
"name": "get_balance",
|
|
"description": "Get account balance",
|
|
"parameters": {"type": "object", "properties": {}}
|
|
},
|
|
{
|
|
"name": "get_price",
|
|
"description": "Get registration price for a TLD",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"tld": {"type": "string", "description": "TLD like com, net, io"}
|
|
},
|
|
"required": ["tld"]
|
|
},
|
|
},
|
|
{
|
|
"name": "register_domain",
|
|
"description": "Register a new domain. WARNING: This will charge your account.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"domain": {"type": "string", "description": "Domain to register (e.g., example.com)"},
|
|
"years": {"type": "integer", "default": 1, "description": "Registration period (1-10 years)"}
|
|
},
|
|
"required": ["domain"]
|
|
}
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
|
|
|
|
# ===== Domain Registration =====
|
|
|
|
class DomainRegisterRequest(BaseModel):
|
|
domain: str
|
|
years: int = 1
|
|
add_whois_guard: bool = True
|
|
telegram_id: Optional[str] = None # 예치금 결제 시 필수
|
|
# Registrant info (optional - uses default if not provided)
|
|
first_name: Optional[str] = None
|
|
last_name: Optional[str] = None
|
|
organization: Optional[str] = None
|
|
address1: Optional[str] = None
|
|
address2: Optional[str] = None
|
|
city: Optional[str] = None
|
|
state_province: Optional[str] = None
|
|
postal_code: Optional[str] = None
|
|
country: Optional[str] = None
|
|
phone: Optional[str] = None
|
|
email: Optional[str] = None
|
|
|
|
|
|
# Default registrant info from environment
|
|
def get_default_registrant() -> RegistrantInfo:
|
|
return RegistrantInfo(
|
|
first_name=os.getenv("REGISTRANT_FIRST_NAME", ""),
|
|
last_name=os.getenv("REGISTRANT_LAST_NAME", ""),
|
|
organization=os.getenv("REGISTRANT_ORGANIZATION", ""),
|
|
address1=os.getenv("REGISTRANT_ADDRESS1", ""),
|
|
address2=os.getenv("REGISTRANT_ADDRESS2", ""),
|
|
city=os.getenv("REGISTRANT_CITY", ""),
|
|
state_province=os.getenv("REGISTRANT_STATE_PROVINCE", ""),
|
|
postal_code=os.getenv("REGISTRANT_POSTAL_CODE", ""),
|
|
country=os.getenv("REGISTRANT_COUNTRY", "JP"),
|
|
phone=os.getenv("REGISTRANT_PHONE", ""),
|
|
email=os.getenv("REGISTRANT_EMAIL", ""),
|
|
)
|
|
|
|
|
|
@app.post("/domains/register")
|
|
async def register_domain(req: DomainRegisterRequest):
|
|
"""
|
|
Register a new domain with deposit balance check.
|
|
|
|
- telegram_id: Required for deposit payment
|
|
- Checks balance before registration
|
|
- Deducts from deposit after successful registration
|
|
"""
|
|
from namecheap import NamecheapError
|
|
from deposit_api import get_balance, deduct_balance
|
|
from db import get_prices, init_db
|
|
|
|
# telegram_id 필수 체크
|
|
if not req.telegram_id:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="telegram_id is required for domain registration"
|
|
)
|
|
|
|
# 1. 도메인 가격 조회
|
|
init_db()
|
|
tld = req.domain.rsplit(".", 1)[-1] if "." in req.domain else req.domain
|
|
prices = get_prices()
|
|
price_info = next((p for p in prices if p["tld"] == tld), None)
|
|
|
|
if not price_info:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"TLD '{tld}' pricing not found"
|
|
)
|
|
|
|
price_krw = price_info["krw"] * req.years
|
|
|
|
# 2. 예치금 잔액 확인
|
|
balance_result = await get_balance(req.telegram_id)
|
|
if "error" in balance_result:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"잔액 조회 실패: {balance_result['error']}"
|
|
)
|
|
|
|
current_balance = balance_result.get("balance", 0)
|
|
if current_balance < price_krw:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"잔액 부족: 현재 {current_balance:,}원, 필요 {price_krw:,}원"
|
|
)
|
|
|
|
# 3. Registrant 정보 준비
|
|
default = get_default_registrant()
|
|
registrant = RegistrantInfo(
|
|
first_name=req.first_name or default.first_name,
|
|
last_name=req.last_name or default.last_name,
|
|
organization=req.organization or default.organization,
|
|
address1=req.address1 or default.address1,
|
|
address2=req.address2 or default.address2,
|
|
city=req.city or default.city,
|
|
state_province=req.state_province or default.state_province,
|
|
postal_code=req.postal_code or default.postal_code,
|
|
country=req.country or default.country,
|
|
phone=req.phone or default.phone,
|
|
email=req.email or default.email,
|
|
)
|
|
|
|
# 필수 필드 체크
|
|
required = ["first_name", "last_name", "address1", "city", "state_province",
|
|
"postal_code", "country", "phone", "email"]
|
|
missing = [f for f in required if not getattr(registrant, f)]
|
|
if missing:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Missing required registrant fields: {', '.join(missing)}. Set REGISTRANT_* env vars."
|
|
)
|
|
|
|
# 4. 도메인 등록
|
|
try:
|
|
result = api.domains_create(
|
|
domain=req.domain,
|
|
registrant=registrant,
|
|
years=req.years,
|
|
add_whois_guard=req.add_whois_guard,
|
|
)
|
|
except NamecheapError as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
# 5. 등록 성공 시 예치금 차감
|
|
if result.get("registered"):
|
|
deduct_result = await deduct_balance(
|
|
telegram_id=req.telegram_id,
|
|
amount=price_krw,
|
|
reason=f"도메인 등록: {req.domain} ({req.years}년)"
|
|
)
|
|
|
|
if "error" in deduct_result:
|
|
# 차감 실패 시 경고 (도메인은 이미 등록됨)
|
|
result["warning"] = f"도메인은 등록되었으나 예치금 차감 실패: {deduct_result['error']}"
|
|
else:
|
|
result["deposit_deducted"] = price_krw
|
|
result["new_balance"] = deduct_result.get("new_balance", 0)
|
|
|
|
return result
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|