from odoo import models, fields, api, _ from odoo.exceptions import UserError import requests import urllib3 import ipaddress class VPSServer(models.Model): _name = 'vps.server' _description = 'VPS Server' _inherit = ['portal.mixin', 'mail.thread', 'mail.activity.mixin'] name = fields.Char(string='Server Name', required=True, tracking=True) ipv4_address = fields.Text(string='IPv4 Addresses', tracking=True) ipv6_address = fields.Text(string='IPv6 Addresses', tracking=True) cpu = fields.Integer(string='CPU Cores', tracking=True) ram = fields.Float(string='RAM (GB)', tracking=True) storage = fields.Float(string='Storage (GB)', tracking=True) os = fields.Char(string='Operating System', tracking=True) customer_id = fields.Many2one('res.partner', string='Customer', required=True, tracking=True) proxmox_server_id = fields.Many2one('proxmox.server', string='Proxmox Server', tracking=True) ct_id = fields.Integer(string='CT ID (LXC)', tracking=True) vm_id = fields.Integer(string='VM ID (QEMU)', tracking=True) state = fields.Selection([ ('running', 'Running'), ('stopped', 'Stopped'), ('unknown', 'Unknown') ], string='State', default='unknown', tracking=True) def _compute_access_url(self): super()._compute_access_url() for server in self: server.access_url = '/my/vps-servers/%s' % server.id @api.onchange('ct_id', 'vm_id') def _onchange_id(self): if self.ct_id and self.vm_id: raise UserError(_("Please provide either CT ID or VM ID, not both.")) def _proxmox_request(self, method, endpoint, data=None): self.ensure_one() proxmox = self.proxmox_server_id base_url = proxmox.url.rstrip('/') headers = { "Authorization": f"PVEAPIToken={proxmox.api_token_id}={proxmox.api_token_secret}" } url = f"{base_url}/api2/json/{endpoint}" urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) try: response = requests.request(method, url, headers=headers, json=data, verify=False) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise UserError(_("Proxmox API request failed: %s") % str(e)) def action_start(self): for server in self: vm_type = 'lxc' if server.ct_id else 'qemu' vm_id = server.ct_id or server.vm_id self._proxmox_request('POST', f'nodes/pve/{vm_type}/{vm_id}/status/start') self.fetch_state() def action_shutdown(self): for server in self: vm_type = 'lxc' if server.ct_id else 'qemu' vm_id = server.ct_id or server.vm_id self._proxmox_request('POST', f'nodes/pve/{vm_type}/{vm_id}/status/shutdown') self.fetch_state() def action_stop(self): for server in self: vm_type = 'lxc' if server.ct_id else 'qemu' vm_id = server.ct_id or server.vm_id self._proxmox_request('POST', f'nodes/pve/{vm_type}/{vm_id}/status/stop') self.fetch_state() def action_reboot(self): for server in self: vm_type = 'lxc' if server.ct_id else 'qemu' vm_id = server.ct_id or server.vm_id self._proxmox_request('POST', f'nodes/pve/{vm_type}/{vm_id}/status/reboot') self.fetch_state() def fetch_state(self): for server in self: vm_type = 'lxc' if server.ct_id else 'qemu' vm_id = server.ct_id or server.vm_id status = self._proxmox_request('GET', f'nodes/pve/{vm_type}/{vm_id}/status/current') server.state = 'running' if status['data']['status'] == 'running' else 'stopped' def action_fetch_proxmox_data(self): self.ensure_one() if not self.proxmox_server_id: raise UserError(_("Please select a Proxmox server first.")) if not self.ct_id and not self.vm_id: raise UserError(_("Please provide either CT ID or VM ID.")) proxmox = self.proxmox_server_id base_url = proxmox.url.rstrip('/') headers = { "Authorization": f"PVEAPIToken={proxmox.api_token_id}={proxmox.api_token_secret}" } try: urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # Determine if it's a container or VM if self.ct_id: vm_type = 'lxc' vm_id = self.ct_id else: vm_type = 'qemu' vm_id = self.vm_id # Fetch VM/CT status status_response = requests.get( f"{base_url}/api2/json/nodes/pve/{vm_type}/{vm_id}/status/current", verify=False, headers=headers ) status_response.raise_for_status() status_data = status_response.json()['data'] # Fetch VM/CT config config_response = requests.get( f"{base_url}/api2/json/nodes/pve/{vm_type}/{vm_id}/config", verify=False, headers=headers ) config_response.raise_for_status() config_data = config_response.json()['data'] # Update fields hostname = config_data.get('hostname', '') vm_name = f"{vm_type}-{vm_id}" self.name = f"{vm_name} ({hostname})" if hostname else vm_name self.cpu = status_data.get('cpus', 0) self.ram = status_data.get('maxmem', 0) / (1024 * 1024 * 1024) # Convert to GB self.storage = status_data.get('maxdisk', 0) / (1024 * 1024 * 1024) # Convert to GB # Try to get IP addresses ipv4_addresses = [] ipv6_addresses = [] def parse_ip(ip): try: ip_obj = ipaddress.ip_address(ip.split('/')[0]) return str(ip_obj), 'v4' if ip_obj.version == 4 else 'v6' except ValueError: return None, None if vm_type == 'lxc': for key, value in config_data.items(): if key.startswith('net'): ip_configs = value.split(',') for config in ip_configs: if config.startswith('ip=') or config.startswith('ip6='): ip = config.split('=')[1] parsed_ip, ip_type = parse_ip(ip) if parsed_ip: if ip_type == 'v4': ipv4_addresses.append(parsed_ip) else: ipv6_addresses.append(parsed_ip) else: # QEMU for key, value in config_data.items(): if key.startswith('ipconfig'): ips = value.split(',') for ip_config in ips: if '=' in ip_config: ip = ip_config.split('=')[1] parsed_ip, ip_type = parse_ip(ip) if parsed_ip: if ip_type == 'v4': ipv4_addresses.append(parsed_ip) else: ipv6_addresses.append(parsed_ip) self.ipv4_address = ', '.join(ipv4_addresses) if ipv4_addresses else 'No IPv4 found' self.ipv6_address = ', '.join(ipv6_addresses) if ipv6_addresses else 'No IPv6 found' # Try to get OS info if vm_type == 'lxc': self.os = config_data.get('ostype', 'Unknown') else: # QEMU self.os = config_data.get('ostype', 'Unknown') self.fetch_state() return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'title': _('Data Fetched'), 'message': _('Successfully fetched data from Proxmox server.'), 'sticky': False, 'type': 'success', } } except requests.exceptions.RequestException as e: raise UserError(_("Failed to fetch data from Proxmox server: %s") % str(e)) @api.model def create(self, vals): record = super(VPSServer, self).create(vals) record.message_subscribe(partner_ids=[record.customer_id.id]) return record def write(self, vals): res = super(VPSServer, self).write(vals) if 'customer_id' in vals: self.message_subscribe(partner_ids=[vals['customer_id']]) return res def unlink(self): for record in self: record.message_unsubscribe(partner_ids=[record.customer_id.id]) return super(VPSServer, self).unlink()