arvan-cli/main.py

#!/bin/python3 

import requests
import argparse
import json

api_key = ""


headers = {
    "Authorization": api_key,
    "Accept": "application/json"
}

def get_all_domains():

    domains_list = {}

    request_domains = requests.get('https://napi.arvancloud.ir/cdn/4.0/domains',headers=headers)

    domains_data = request_domains.json()['data']

    for domain in domains_data:
        domain_name = domain['domain']
        domain_status = domain['status']
        domains_list[domain_name] = {
            'name': domain_name,
            'status': domain_status
        }
    
    return json.dumps(domains_list)

def get_all_domain_records(target_domain):
    domains_records = {}
    request_domains_records = requests.get(f'https://napi.arvancloud.ir/cdn/4.0/domains/{target_domain}/dns-records',headers=headers)
    domains_records_data = request_domains_records.json()['data']
    for record in domains_records_data:
        record_name = record['name']
        record_type = record['type']
        record_ttl = record['ttl']
        
        record_values_list = []

        for value in record['value']:
            if record_type == 'a':
                record_values_list.append({
                    'ip': value['ip'],
                    'port': value['port']
                })

                domains_records[record_name] = {
                    'type': record_type,
                    'ttl': record_ttl,
                    'values': record_values_list
                }

def get_domain_record_id(target_domain,target_record):
    req = requests.get(f'https://napi.arvancloud.ir/cdn/4.0/domains/{target_domain}/dns-records',headers=headers)
    if req.status_code == 200:
       data = req.json()['data']
       for value in data:
           if value['name'] == target_record:
               return value['id']


def create_domain_records(target_domain,metadata):
    req = requests.post(f'https://napi.arvancloud.ir/cdn/4.0/domains/{target_domain}/dns-records',headers=headers,json=metadata)
    if req.status_code == 200:
        print('Records Created ')
    print(req.text)

def remove_domain_record(target_domain,target_record):
    records_id = get_domain_record_id(target_domain,target_record)
    req = requests.delete(f'https://napi.arvancloud.ir/cdn/4.0/domains/{target_domain}/dns-records/{records_id}',headers=headers)
    if req.status_code == 200: 
        print(f'Sub Domain {target_record} Removed On Domain {target_domain}')
    else: 
        print(f'Can Not Remove Sub Domain {target_record} On Domain {target_domain}')

def update_domain_record(target_domain,metadata):
    target_record = metadata['name']
    records_id = get_domain_record_id(target_domain,target_record)
    print(records_id)
    # req = requests.put(f'https://napi.arvancloud.ir/cdn/4.0/domains/{target_domain}/dns-records/{records_id}',headers=headers)



def cli_create(args):
    metadata = {
        "name": args.name,
        "type": args.type,
        "ttl": args.ttl,
        "cloud": args.cloud,
        "value": [{"ip": args.ip}]
    }

    create_domain_records(args.domain, metadata)


def cli_remove(args):
    remove_domain_record(args.domain, args.name)


def cli_update(args):
    metadata = {
        "name": args.name,
        "type": args.type,
        "ttl": args.ttl,
        "cloud": args.cloud,
        "value": [{"ip": args.ip}]
    }

    update_domain_record(args.domain, metadata)


def cli_list_domains(args):
    res = get_all_domains()
    print(res)

def main():
    parser = argparse.ArgumentParser(description="ArvanCloud CDN CLI Tool")
    subparsers = parser.add_subparsers(dest="command")

    # cdn
    cdn_parser = subparsers.add_parser("cdn")
    cdn_subparsers = cdn_parser.add_subparsers(dest="cdn_command")

    # records
    records_parser = cdn_subparsers.add_parser("records")
    records_subparsers = records_parser.add_subparsers(dest="action")

    # create
    create_parser = records_subparsers.add_parser("create")
    create_parser.add_argument("--domain", required=True)
    create_parser.add_argument("--name", required=True)
    create_parser.add_argument("--type", default="a")
    create_parser.add_argument("--ttl", type=int, default=120)
    create_parser.add_argument("--ip", required=True)
    create_parser.add_argument("--cloud", action="store_true")
    create_parser.set_defaults(func=cli_create)

    # remove
    remove_parser = records_subparsers.add_parser("remove")
    remove_parser.add_argument("--domain", required=True)
    remove_parser.add_argument("--name", required=True)
    remove_parser.set_defaults(func=cli_remove)

    # update
    update_parser = records_subparsers.add_parser("update")
    update_parser.add_argument("--domain", required=True)
    update_parser.add_argument("--name", required=True)
    update_parser.add_argument("--type", default="a")
    update_parser.add_argument("--ttl", type=int, default=120)
    update_parser.add_argument("--ip", required=True)
    update_parser.add_argument("--cloud", action="store_true")
    update_parser.set_defaults(func=cli_update)

    # domains list
    domains_parser = cdn_subparsers.add_parser("domains")
    domains_parser.set_defaults(func=cli_list_domains)

    args = parser.parse_args()

    if hasattr(args, "func"):
        args.func(args)
    else:
        parser.print_help()


if __name__ == "__main__":
    main()