2 Commits
0.2.4 ... 0.3

Author SHA1 Message Date
8000779fe8 Added app version checks. 2023-09-06 12:29:32 +02:00
8814127cf8 Added some functionality 2023-09-05 11:36:39 +02:00

104
malias.py
View File

@@ -29,9 +29,27 @@ database = filepath.joinpath('malias.db')
logfile = filepath.joinpath('malias.log')
Path(filepath).mkdir(parents=True, exist_ok=True)
logging.basicConfig(filename=logfile,level=logging.INFO,format='%(message)s')
app_version = '0.2.4'
app_version = '0.3'
def get_latest_release():
req = urllib.request.Request('https://gitlab.pm/api/v1/repos/rune/malias/releases/latest')
req.add_header('Content-Type', 'application/json')
current = urllib.request.urlopen(req)
remote = current.read().decode('utf-8')
remoteData = json.loads(remote)
return remoteData['tag_name']
latest_release = get_latest_release()
def release_check():
if app_version != latest_release:
print('[b]New version available @ [i]https://iurl.no/malias[/b][/i]')
else:
print ('')
def connect_database():
Path(filepath).mkdir(parents=True, exist_ok=True)
conn = None
@@ -99,6 +117,7 @@ def get_settings(kind):
def get_api():
latest_release = get_latest_release()
cursor = conn.cursor()
cursor.execute('SELECT api FROM apikey')
apikey = cursor.fetchone()[0]
@@ -116,7 +135,7 @@ def set_mailserver(server):
logging.info(now + ' - Info : mailcow server updated')
print('Your mail server has been updated.')
conn.commit()
release_check()
def apikey(key):
@@ -126,9 +145,9 @@ def apikey(key):
logging.info(now + ' - Info : API key updated')
print('Your API key has been updated.')
conn.commit()
release_check()
def get_mail_domains():
def get_mail_domains(info):
apikey = get_api()
cursor = conn.cursor()
mail_server = get_settings('mail_server')
@@ -138,18 +157,21 @@ def get_mail_domains():
current = urllib.request.urlopen(req)
remote = current.read().decode('utf-8')
remoteData = json.loads(remote)
total_aliases = 0
i=0
print('\n[b]malias[/b] - All email domains on %s' %(mail_server))
print('==================================================================')
for domains in remoteData:
cursor.execute('SELECT count(*) FROM aliases where alias like ? or goto like ?', ('%'+remoteData[i]['domain_name']+'%','%'+remoteData[i]['domain_name']+'%',))
count = cursor.fetchone()[0]
total_aliases += count
print('%s \t\twith %s aliases' %(remoteData[i]['domain_name'],count))
i+=1
print('\n\nThere is a total of %s domains with %s aliases.' %(str(i),str(total_aliases)))
if info:
total_aliases = 0
i=0
print('\n[b]malias[/b] - All email domains on %s' %(mail_server))
print('==================================================================')
for domains in remoteData:
cursor.execute('SELECT count(*) FROM aliases where alias like ? or goto like ?', ('%'+remoteData[i]['domain_name']+'%','%'+remoteData[i]['domain_name']+'%',))
count = cursor.fetchone()[0]
total_aliases += count
print('%s \t\twith %s aliases' %(remoteData[i]['domain_name'],count))
i+=1
print('\n\nThere is a total of %s domains with %s aliases.' %(str(i),str(total_aliases)))
release_check()
else:
return(remoteData)
def create(alias,to_address):
@@ -175,7 +197,7 @@ def create(alias,to_address):
conn.commit()
logging.info(now + ' - Info : alias %s created for %s on the mailcow instance %s ' %(alias,to_address,server))
print('\n[b]Info[/b] : alias %s created for %s on the mailcow instance %s \n' %(alias,to_address,server))
release_check()
def delete_alias(alias):
@@ -206,6 +228,8 @@ def delete_alias(alias):
else:
print('\n[b]Error[/b] : The alias %s not found')
release_check()
def copy_data():
@@ -232,6 +256,7 @@ def copy_data():
else:
print('\n[b]Info[/b] : aliases alreday imported from the mailcow instance %s to local DB\n\n[i]Updating with any missing aliases![/i]' %(mail_server))
update_data()
release_check()
def update_data():
@@ -266,7 +291,6 @@ def update_data():
print('\n[b]Info[/b] : No missing aliases from local DB \n')
def checklist(alias):
alias_exist = None
apikey = get_api()
@@ -317,8 +341,7 @@ def list_alias():
print(the_alias + '\tgoes to\t\t' + the_goto)
i=i+1
print('\n\nTotal number of aliases %s on instance [b]%s[/b] and %s on [b]local DB[/b].' %(str(i),mail_server,str(l)))
release_check()
def alias_id(alias):
apikey = get_api()
@@ -365,7 +388,6 @@ def check_local_db(alias_id):
def search(alias):
apikey = get_api()
mail_server = get_settings('mail_server')
@@ -377,13 +399,14 @@ def search(alias):
cursor.execute('SELECT * from aliases where alias like ? or goto like ?',(search_term,search_term,))
remoteData = cursor.fetchall()
i = 0
print('\nAliases on %s that contains %s' %(mail_server,alias))
print('\nAliases on %s that contains [b]%s[/b]' %(mail_server,alias))
print('=================================================================')
for search in remoteData:
the_alias = remoteData[i][1].ljust(20,' ')
print(the_alias + '\tgoes to\t\t' + remoteData[i][2])
i=i+1
print('\n\nData from local DB')
print('\n\nData from local DB\n Run [i]malias -c to get any aliases on the server but [b]not[/b] in the local DB![/i]')
release_check()
else:
req = urllib.request.Request('https://'+mail_server+'/api/v1/get/alias/all')
req.add_header('Content-Type', 'application/json')
@@ -392,15 +415,16 @@ def search(alias):
remote = current.read().decode('utf-8')
remoteData = json.loads(remote)
i = 0
print('\nAliases on %s that contains %s' %(mail_server,alias))
print('\nAliases on %s that contains [b]%s[/b]' %(mail_server,alias))
print('=================================================')
for search in remoteData:
if alias in remoteData[i]['address'] or alias in remoteData[i]['goto']:
print(remoteData[i]['address'] + '\tgoes to\t\t' + remoteData[i]['goto'])
i=i+1
print('\n\nData from server')
release_check()
def get_mailcow_version():
apikey = get_api()
mail_server = get_settings('mail_server')
@@ -418,15 +442,15 @@ def get_mailcow_version():
tags = remote_heads.splitlines()
for x in tags:
string = x.rsplit('/',2)
if remoteData['version'] != string[2]:
versionInfo = 'Your Mailcow version is %s and the latest is %s' %(remoteData['version'], string[2])
else:
versionInfo = 'You have the latest Mailcow version %s' %remoteData['version']
return (versionInfo)
def show_current_info():
API = get_api()
mail_server = get_settings('mail_server')
@@ -440,18 +464,30 @@ def show_current_info():
aliases_server = number_of_aliases_on_server()
alias_db = number_of_aliases_in_db()
mailcow_version = get_mailcow_version()
mail_domains = get_mail_domains(False)
domain = ""
i=0
for domains in mail_domains:
if i!=0:
domain = domain + ', ' + str(mail_domains[i]['domain_name'])
else:
domain = domain + str(mail_domains[i]['domain_name'])
i+=1
print('\n[b]malias[/b] - Manage aliases on mailcow Instance.')
print('===================================================')
print('API key : [b]%s[/b]' % (API))
print('mailcow Instance : [b]%s[/b]' % (mail_server))
print('Mailcow Instance : [b]%s[/b]' % (mail_server))
print('Active domains : [b]%s[/b]' % (domain))
print('Mailcow version : [b]%s[/b]' % (mailcow_version))
print('Logfile : [b]%s[/b]' % (logfile))
print('Databse : [b]%s[b]' % (database))
print('Aliases on server : [b]%s[/b]' % (aliases_server))
print('Aliases in DB : [b]%s[/b]' % (alias_db))
print('')
print('App version : [b]%s[/b] (https://iurl.no/malias)' % (app_version))
if app_version != latest_release:
print('App version : [b]%s[/b] a new version (%s) is available @ https://iurl.no/malias' % (app_version,latest_release))
else:
print('App version : [b]%s[/b]' % (app_version))
print('')
@@ -512,9 +548,9 @@ elif args['copy']:
elif args['list']:
list_alias()
elif args['domains']:
get_mail_domains()
get_mail_domains(True)
else:
#get_mailcow_version()
# get_api()
print('\n\nEh, sorry! I need something more to help you! If you write [b]malias -h[/b] I\'ll show a help screen to get you going!!!\n\n\n')