connor/dotfiles
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rebuild dotfiles as one branch with per-host layers

Browse Source 
Replaces the previous repo, which had split into two histories that never
met (mainframe on a dead GitLab remote, the laptops on Gitea) with 146
dirty files across three machines and the NAS never enrolled at all.

Branch-per-machine is gone. One main, with host differences expressed as
small files under hosts/<hostname>/ rather than as branches, so there is
nothing to merge. The reconciled zsh layer reduces 15-33 line forks to
1-7 effective lines per host; distro differences (oh-my-zsh prefix,
syntax-highlighting path, fd vs fdfind) are probed in common/ instead.

Fresh history: the old one carried six plaintext credentials, 45 MB of
mail caches, browser caches and vendored binaries. 5,096 tracked files
and 144 MB become 462 files and 2.6 MB. The .gitignore is now an
allowlist, which is what keeps that true.

Root cause of the rot: ~/.local/bin was a symlink to scripts/ with GOPATH
inside it, so every go install wrote into version control (2.2 GB on the
work laptop). PATH now points at the repo instead of the reverse.

Also: Hyprland replaces sway and is sourced in two halves so $browser is
defined before use; singleton automations carry ConditionHost= alongside
host-layer-only placement; ddns moves from cron to a guarded timer;
package manifests and pkg-snapshot/pkg-restore replace the X11-era
install_scripts/; networkmanager-dmenu added to system76 (the binding
always existed, the package never did).
This commit is contained in:
connor
2026-09-14 14:24:37 -04:00
commit e6644d0616
462 changed files with 23524 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hosts/mainframe/bin/ddns
Executable
+665
View File
@@ -0,0 +1,665 @@
#!/usr/bin/python3
import sqlite3
from pathlib import Path
from sqlite3 import Error
import urllib.request
import json
import logging
import argparse
import requests
import os
import time
from datetime import datetime
from string import ascii_letters, digits
from rich import print
from argparse import RawTextHelpFormatter
homefilepath = Path.home()
filepath = homefilepath.joinpath('.config/ddns')
database = filepath.joinpath('ddns.db')
logfile = filepath.joinpath('ddns.log')
logging.basicConfig(filename=logfile,level=logging.INFO,format='%(message)s')
app_version = '0.5.2'
def get_ip():
cursor = conn.cursor()
cursor.execute('SELECT COUNT(ip4_server) FROM ipservers')
count = cursor.fetchone()[0]
if count != 0:
cursor.execute('SELECT ip4_server from ipservers')
server = cursor.fetchone()
return server[0]
else:
return None
def connect_database():
Path(filepath).mkdir(parents=True, exist_ok=True)
conn = None
try:
conn = sqlite3.connect(database)
except Error as e:
logging.error(time.strftime("%Y-%m-%d %H:%M") + ' - Error : ' + str(e))
print(e)
finally:
if conn:
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS apikey
(id integer NOT NULL PRIMARY KEY,
api text NOT NULL)''')
c.execute('''CREATE TABLE IF NOT EXISTS ipservers
(id integer NOT NULL PRIMARY KEY,
ip4_server text NOT NULL,
ip6_server text)''')
c.execute('''CREATE TABLE IF NOT EXISTS domains
(id integer PRIMARY KEY,
name text NOT NULL)''')
c.execute('''CREATE TABLE IF NOT EXISTS subdomains
(id integer PRIMARY KEY,
main_id integer NOT NULL,
name text NOT NULL,
current_ip4 text NOT NULL,
current_ip6 text NULL)''')
return conn
def get_api():
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM apikey')
count = cursor.fetchone()[0]
if count == 0:
return None
else:
cursor.execute('SELECT * FROM apikey')
rows = cursor.fetchone()
return rows[1]
def api(api_value):
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM apikey')
count = cursor.fetchone()[0]
if count == 0:
cursor.execute('INSERT INTO apikey values(?,?)', (1, api_value))
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : API key added')
print('Your API key has been added.')
else:
cursor.execute('UPDATE apikey SET api = ? WHERE id = 1',(api_value,))
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : API key updated')
print('Your API key has been updated.')
conn.commit()
def add_domian(domain):
apikey = get_api()
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM domains WHERE name like ?',(domain,))
count = cursor.fetchone()[0]
if count != 0:
print('[red]Error:[/red] Domain name (%s) already in database!' % (domain))
else:
if apikey != None:
headers = {'Authorization': 'Bearer ' + apikey, "Content-Type": "application/json"}
response = requests.get('https://api.digitalocean.com/v2/domains/' + domain, headers=headers)
response_data = response.json()
if 'id' in response_data:
print('[red]Error: [/red]The domain does not exist in your DigitalOcean account.\nPlease add the domain from your control panel [b]https://cloud.digitalocean.com/networking/domains/[/b]')
else:
cursor.execute('INSERT INTO domains values(?,?)', (None, domain,))
print('The domain [b]%s[/b] has been added to the DB' % (domain))
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : Domain %s added' %(domain))
conn.commit()
def add_subdomain(domain):
now = datetime.now().strftime("%Y-%m-%d %H:%M")
if set(domain).difference(ascii_letters + '.' + digits + '-' + '@'):
print('[red]Error:[/red] Give the domain name in simple form e.g. [b]test.domain.com[/b]')
else:
parts = domain.split('.')
if len(parts) > 3:
top = parts[1] + '.' + parts[2] + '.' + parts[3]
sub = parts[0]
else:
sub = parts[0]
top = parts[1] + '.' + parts[2]
apikey = get_api()
if apikey == None:
print("[red]Error:[/red] Missing APIkey. Please add one!")
else:
ip = get_ip()
if ip == None or 'urlopen error' in ip:
print('[red]Error:[/red] Failed to get public IP. Do you have a typo in your URI? [red]Error %s.[/red]' % (ip))
else:
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM domains WHERE name like ?',(top,))
count = cursor.fetchone()[0]
if count == 0:
print('[red]Error:[/red] Top domain [bold]%s[/bold] does not exist in the DB. Please add it with [i]ddns -t %s[/i].' % (top,top))
else:
cursor.execute('SELECT id,name FROM domains WHERE name LIKE ?',(top,))
topdomain = cursor.fetchone()
topdomain_id = topdomain[0]
topdomain_name = topdomain[1]
cursor.execute('SELECT count(*) FROM subdomains WHERE main_id LIKE ? AND name like ?',(topdomain_id,sub,))
count = cursor.fetchone()[0]
if count != 0:
print('[red]Error:[/red] [bold]%s[/bold] already exists.' % (domain))
else:
data = {'name': sub,'data': ip,'type': "A",'ttl': 3600}
headers = {'Authorization': 'Bearer ' + apikey, "Content-Type": "application/json"}
response = requests.post('https://api.digitalocean.com/v2/domains/' + top + '/records',
data=json.dumps(data), headers=headers)
if str(response) == '<Response [201]>':
if response != 'Fail':
response_data = response.json()
domainid = str(response_data['domain_record']['id'])
cursor.execute('INSERT INTO subdomains values(?,?,?,?,?,?,?,?,?)',(domainid,topdomain_id,sub,ip,None,now,now,now,1,))
conn.commit()
print('The domain %s has been added.' % (domain))
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : subdomain %s added'%(domain))
else:
print(f"Failed to add subdomain: {response}, {response.json()}")
return '[red]Error: %s [/red]' % (str(response))
def remove_subdomain(domain):
if set(domain).difference(ascii_letters + '.' + digits + '-' + '@'):
print('[red]Error:[/red] Give the domain name in simple form e.g. [b]test.domain.com[/b]')
else:
parts = domain.split('.')
if len(parts) > 3:
top = parts[1] + '.' + parts[2] + '.' + parts[3]
sub = parts[0]
else:
sub = parts[0]
top = parts[1] + '.' + parts[2]
longtop=sub+'.'+top
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM domains WHERE name like ? or name like ?',(top,longtop,))
count = cursor.fetchone()[0]
if count == 0:
print('[red]Error:[/red] Top domain [bold]%s[/bold] does not exist in the DB. So I\'m giving up!.' % (top))
else:
cursor.execute('SELECT COUNT(*) FROM subdomains WHERE name like ? and main_id=(SELECT id from domains WHERE name like ? or name like ?)',(sub,top,longtop,))
count = cursor.fetchone()[0]
if count == 0:
print('[red]Error:[/red] Domain [bold]%s[/bold] does not exist in the DB. So I\'m giving up!.' % (domain))
else:
apikey = get_api()
if apikey == None:
print("[red]Error:[/red] Missing APIkey. Please add one!")
else:
cursor.execute('SELECT id FROM subdomains WHERE name like ? and main_id=(SELECT id from domains WHERE name like ? or name like ?)',(sub,top,longtop,))
subdomain_id = str(cursor.fetchone()[0])
headers = {'Authorization': 'Bearer ' + apikey, "Content-Type": "application/json"}
response = requests.delete('https://api.digitalocean.com/v2/domains/'+top+'/records/' + subdomain_id, headers=headers)
if str(response) == '<Response [204]>':
cursor.execute('DELETE from subdomains where id=?',(subdomain_id,))
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : Subdomain %s removed' %(domain))
conn.commit()
else:
print('[red]Error: [/red]An error occurred! Please try again later!')
def edit_subdomain(domain):
if set(domain).difference(ascii_letters + '.' + digits + '-' + '@'):
print('[red]Error:[/red] Give the domain name in simple form e.g. [b]test.domain.com[/b]')
else:
parts = domain.split('.')
if len(parts) > 3:
top = parts[1] + '.' + parts[2] + '.' + parts[3]
sub = parts[0]
else:
sub = parts[0]
top = parts[1] + '.' + parts[2]
longtop=sub+'.'+top
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM domains WHERE name like ? or name like ?',(top,longtop,))
count = cursor.fetchone()[0]
if count == 0:
print('[red]Error:[/red] Top domain [bold]%s[/bold] does not exist in the DB. So I\'m giving up!.' % (top))
else:
cursor.execute('SELECT COUNT(*) FROM subdomains WHERE name like ? and main_id=(SELECT id from domains WHERE name like ? or name like ?)',(sub,top,longtop))
count = cursor.fetchone()[0]
if count == 0:
print('[red]Error:[/red] Domain [bold]%s[/bold] does not exist in the DB. So I\'m giving up!.' % (domain))
else:
apikey = get_api()
if apikey == None:
print("[red]Error:[/red] Missing APIkey. Please add one!")
else:
cursor.execute('SELECT id,active FROM subdomains WHERE name like ? and main_id=(SELECT id from domains WHERE name like ? or name like ?)',(sub,top,longtop))
domain_info = cursor.fetchone()
subdomain_id = str(domain_info[0])
status = domain_info[1]
if status == 1:
status = 0
else:
status = 1
cursor.execute('UPDATE subdomains SET active = ? WHERE id = ?',(status,subdomain_id,))
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : Status for domain %s changed' %(domain))
print('Status for domain %s changed' %(domain))
conn.commit()
def show_all_top_domains():
cursor = conn.cursor()
apikey = get_api()
if apikey != None:
req = urllib.request.Request('https://api.digitalocean.com/v2/domains/?per_page=200')
req.add_header('Content-Type', 'application/json')
req.add_header('Authorization', 'Bearer ' + apikey)
current = urllib.request.urlopen(req)
remote = current.read().decode('utf-8')
remoteData = json.loads(remote)
print('Domains in database are marked with a [*]')
print('================================================')
for k in remoteData["domains"]:
cursor.execute('SELECT COUNT(*) FROM domains WHERE name like ?',(k['name'],))
count = cursor.fetchone()[0]
if count != 0:
print('Name : [bold]'+k['name']+ ' [*][/bold]')
else:
print('Name : '+k['name'])
else:
print("[red]Error:[/red] Missing APIkey. Please add one!")
def list_sub_domains(domain):
apikey = get_api()
cursor = conn.cursor()
if apikey == None:
print("[red]Error:[/red] Missing APIkey. Please add one!")
else:
cursor.execute('SELECT COUNT(*) FROM domains WHERE name LIKE ?',(domain,))
count = cursor.fetchone()[0]
if count == 0:
print("[red]Error: [/red]No such domain. Check spelling or use ddns -d to show all top domains.")
else:
print('\n\nCurrent sub domains for [b]%s[/b]\n\n' % (domain))
print('Domain\t\t\t\tCreated\t\t\tUpdated\t\t\tChecked\t\t\tActive')
print('==================================================================================================================')
cursor.execute('SELECT id FROM domains WHERE name LIKE ?', (domain,))
topdomain_id = cursor.fetchone()[0]
cursor.execute('SELECT COUNT(*) FROM subdomains WHERE main_id LIKE ?',(topdomain_id,))
count = cursor.fetchone()[0]
if count == 0:
print('[red]Error:[/red] No sub domains for [b]%s[/b]' % (domain))
else:
cursor.execute('SELECT name,last_updated,last_checked,created,active FROM subdomains WHERE main_id LIKE ?',(topdomain_id,) )
subdomains = cursor.fetchall()
for i in subdomains:
if i[4] == 1:
active = 'True'
else:
active = 'False'
topdomain = i[0]+'.'+domain
topdomain = "{:<25}".format(topdomain)
print(topdomain+'\t'+i[3]+'\t'+i[1]+'\t'+i[2]+'\t'+active)
print('\n')
def list_do_sub_domains(domain):
apikey = get_api()
cursor = conn.cursor()
if apikey == None:
print("[red]Error:[/red] Missing APIkey. Please add one!")
else:
req = urllib.request.Request('https://api.digitalocean.com/v2/domains/'+domain+'/records?type="A"/?per_page=200')
req.add_header('Content-Type', 'application/json')
req.add_header('Authorization', 'Bearer ' + apikey)
current = urllib.request.urlopen(req)
remote = current.read().decode('utf-8')
remoteData = json.loads(remote)
print('Domains in your DigitalOcean account not in ddns DB for [b]%s[/b]' % (domain))
print('===================================================================')
for k in remoteData["domain_records"]:
if k['type'] == 'A':
cursor.execute('SELECT COUNT(*) FROM subdomains WHERE id like ?',(str(k['id']),))
count = cursor.fetchone()[0]
if count == 0:
print(k['name']+'.'+domain+'\t\tID : '+str(k['id']))
def domaininfo(domain):
apikey = get_api()
local_ip = get_ip()
cursor = conn.cursor()
if set(domain).difference(ascii_letters + '.' + digits + '@' + '-'):
print('[red]Error:[/red]. Give the domain name in simple form e.g. [bold]test.domain.com[/bold]')
else:
parts = domain.split('.')
if len(parts) > 3:
top = parts[1] + '.' + parts[2] + '.' + parts[3]
sub = parts[0]
else:
sub = parts[0]
top = parts[1] + '.' + parts[2]
cursor.execute('SELECT id FROM domains WHERE name like ?', (top,))
domainid = cursor.fetchone()[0]
cursor.execute('SELECT * FROM subdomains WHERE main_id like ?', (domainid,))
domains = cursor.fetchall()
if local_ip != domains[0][3]:
localip = '[red]%s[/red]' % (local_ip)
else:
localip = local_ip
print ('The domain [bold]%s[/bold] has the IP [bold]%s[/bold]. Your public IP is [bold]%s[/bold]' % (domain,domains[0][3],localip))
def show_current_info():
ipserver = None
API = get_api()
cursor = conn.cursor()
cursor.execute('SELECT COUNT(ip4_server) FROM ipservers')
count = cursor.fetchone()[0]
if count == 0:
ipserver = '[red]Error:[/red] No IP resolvers in DB'
else:
cursor.execute('SELECT * FROM ipservers')
ipservers = cursor.fetchall()
ip4server = ipservers[0][1]
ip6server = ipservers[0][2]
if API == None:
API = '[red]Error:[/red] API key not stored in DB'
cursor.execute('SELECT COUNT(*) FROM domains')
topdomains = cursor.fetchone()[0]
cursor.execute('SELECT COUNT(*) FROM subdomains')
subdomains = cursor.fetchone()[0]
print('\n[b]ddns[/b] - a DigitalOcean dynamic DNS solution.')
print('===================================================')
print('API key : [b]%s[/b]' % (API))
print('IP v4 resolver : [b]%s[/b]' % (ip4server))
print('IP v6 resolver : [b]%s[/b]' % (ip6server))
print('Logfile : [b]%s[/b]' % (logfile))
print('Top domains : [b]%s[/b]' % (topdomains))
print('sub domains : [b]%s[/b]' % (subdomains))
print('')
print('App version : [b]%s[/b] (https://gitlab.pm/rune/ddns)' % (app_version))
print('')
print('[i]IPv6 is not supported and not listed here.[/i]')
def ip_server(ipserver, ip_type):
cursor = conn.cursor()
if ip_type == '4':
cursor.execute('SELECT COUNT(ip4_server) FROM ipservers')
count = cursor.fetchone()[0]
if count == 0:
cursor.execute('INSERT INTO ipservers values(?,?,?)', (None, ipserver,None))
conn.commit()
print('New IP resolver (%s) for ipv%s added.' % (ipserver, ip_type))
else:
cursor.execute('UPDATE ipservers SET ip4_server = ? WHERE id = 1',(ipserver,))
print('IP resolver (%s) for ipv%s updated.' % (ipserver, ip_type))
logging.info(time.strftime("%Y-%m-%d %H:%M")+' - Info : IP resolver (%s) for ipv%s updated.' % (ipserver, ip_type))
conn.commit()
elif ip_type == '6':
cursor.execute('SELECT COUNT(ip6_server) FROM ipservers')
count = cursor.fetchone()[0]
if count == 0:
cursor.execute('INSERT INTO ipservers values(?,?,?)', (None, None,ipserver))
conn.commit()
print('New IP resolver (%s) for ipv%s added. \n\r This IP version is not supported.' % (ipserver, ip_type))
else:
cursor.execute('UPDATE ipservers SET ip6_server = ? WHERE id = 1',(ipserver,))
print('IP resolver (%s) for ipv%s updated. \n\r This IP version is not supported.' % (ipserver, ip_type))
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : IP resolver (%s) for ipv%s updated.' % (ipserver, ip_type))
conn.commit()
def updateip(force):
apikey = get_api()
current_ip = get_ip()
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM subdomains')
count = cursor.fetchone()[0]
now = datetime.now().strftime("%d-%m-%Y %H:%M")
updated = None
if count == 0:
print('[red]Error: [/red]There are no dynamic domains active.'\
' Start by adding a new domain with [i]ddns -s test.example.com[/i]')
else:
cursor.execute('SELECT id,active FROM subdomains')
rows = cursor.fetchall()
for i in rows:
cursor.execute('SELECT name FROM domains WHERE id like (SELECT main_id from subdomains WHERE id = ?)',(i[0],))
domain_info = cursor.fetchone()
domain_name = str(domain_info[0])
domain_status = i[1]
subdomain_id = str(i[0])
# Chek if an update is required
if domain_status == 1:
req = urllib.request.Request('https://api.digitalocean.com/v2/domains/' + domain_name + '/records/' + subdomain_id)
req.add_header('Content-Type', 'application/json')
req.add_header('Authorization', 'Bearer ' + apikey)
# A dead record id (404) must not abort the whole run: log it,
# leave the row alone, and carry on with the other subdomains.
try:
current = urllib.request.urlopen(req)
remote = current.read().decode('utf-8')
except Exception as e:
cursor.execute('SELECT name FROM subdomains WHERE id = ?',(subdomain_id,))
row = cursor.fetchone()
sub_name = row[0] if row else '?'
logging.error(time.strftime("%Y-%m-%d %H:%M")
+ ' - Error : could not read record for %s.%s (id %s) : %s'
% (sub_name, domain_name, subdomain_id, e))
print('[red]Error:[/red] could not read record for %s.%s (id %s): %s'
% (sub_name, domain_name, subdomain_id, e))
continue
remoteData = json.loads(remote)
remoteIP4 = remoteData['domain_record']['data']
domainname = str(remoteData['domain_record']['name'])
if remoteIP4 != current_ip or force == True and domain_status == 1:
updated = True
data = {'type': 'A', 'data': current_ip}
headers = {'Authorization': 'Bearer ' + apikey, "Content-Type": "application/json"}
response = requests.patch('https://api.digitalocean.com/v2/domains/'+domain_name+'/records/' + subdomain_id, data=json.dumps(data), headers=headers)
if str(response) != '<Response [200]>':
logging.error(time.strftime("%Y-%m-%d %H:%M")+' - Error updating ('+str(domain_name)+') : ' + str(response.content))
else:
cursor.execute('UPDATE subdomains SET current_ip4=? WHERE id = ?',(current_ip,subdomain_id,))
cursor.execute('UPDATE subdomains SET last_updated=? WHERE id = ?',(now,subdomain_id,))
cursor.execute('UPDATE subdomains SET last_checked=? WHERE id = ?',(now,subdomain_id,))
conn.commit()
else:
cursor.execute('UPDATE subdomains SET last_checked=? WHERE id = ?',(now,subdomain_id,))
conn.commit()
if updated == None:
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : No updated necessary')
else:
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : Updates done. Use ddns -l domain.com to check domain')
def local_add_subdomain(domain,domainid):
now = datetime.now().strftime("%Y-%m-%d %H:%M")
if set(domain).difference(ascii_letters + '.' + digits + '-' + '@'):
print('[red]Error:[/red] Give the domain name in simple form e.g. [b]test.domain.com[/b]')
else:
parts = domain.split('.')
if len(parts) > 3:
top = parts[1] + '.' + parts[2] + '.' + parts[3]
sub = parts[0]
else:
sub = parts[0]
top = parts[1] + '.' + parts[2]
apikey = get_api()
longtop=sub+'.'+top
if apikey == None:
print("[red]Error:[/red] Missing APIkey. Please add one!")
else:
ip = get_ip()
if ip == None or 'urlopen error' in ip:
print('[red]Error:[/red] Failed to get public IP. Do you have a typo in your URI? [red]Error %s.[/red]' % (ip))
else:
cursor = conn.cursor()
cursor.execute('SELECT COUNT(*) FROM domains WHERE name like ? or name like ?',(top,longtop,))
count = cursor.fetchone()[0]
if count == 0:
print('[red]Error:[/red] Top domain [bold]%s[/bold] does not exist in the DB. Please add it with [i]ddns -t %s[/i].' % (top,top))
else:
cursor.execute('SELECT id FROM domains WHERE name LIKE ? or name like ?',(top,longtop,))
topdomain_id = cursor.fetchone()
topdomain_id = topdomain_id[0]
cursor.execute('SELECT count(*) FROM subdomains WHERE main_id LIKE ? AND name like ?',(topdomain_id,sub,))
count = cursor.fetchone()[0]
if count != 0:
print('[red]Error:[/red] [bold]%s[/bold] already exists.' % (domain))
else:
cursor.execute('INSERT INTO subdomains values(?,?,?,?,?,?,?,?,?)',(domainid,topdomain_id,sub,ip,None,now,now,now,1,))
conn.commit()
print('The domain %s has been added.' % (domain))
def show_log():
log_file = open(logfile, 'r')
content = log_file.read()
print(content)
log_file.close()
def updatedb():
# Update DB with new column 20.03.23
# Add last updated field for subdomains
new_column = 'last_updated'
info = conn.execute("PRAGMA table_info('subdomains')").fetchall()
if not any(new_column in word for word in info):
add_column = "ALTER TABLE subdomains ADD COLUMN last_updated text default 'N/A'"
conn.execute(add_column)
conn.commit()
logging.info(time.strftime("%Y-%m-%d %H:%M")+' - Info : Database updated')
new_column = 'last_checked'
info = conn.execute("PRAGMA table_info('subdomains')").fetchall()
if not any(new_column in word for word in info):
add_column = "ALTER TABLE subdomains ADD COLUMN last_checked text default 'N/A'"
conn.execute(add_column)
conn.commit()
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : Database updated')
new_column = 'created'
info = conn.execute("PRAGMA table_info('subdomains')").fetchall()
if not any(new_column in word for word in info):
add_column = "ALTER TABLE subdomains ADD COLUMN created text default '[b]Unknown Info[/b]'"
conn.execute(add_column)
conn.commit()
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : Database updated')
new_column = 'active'
info = conn.execute("PRAGMA table_info('subdomains')").fetchall()
if not any(new_column in word for word in info):
add_column = "ALTER TABLE subdomains ADD COLUMN active integer default 1"
conn.execute(add_column)
conn.commit()
logging.info(time.strftime("%Y-%m-%d %H:%M") + ' - Info : Database updated')
# Commandline arguments
conn = connect_database()
updatedb()
parser = argparse.ArgumentParser(prog='ddns',
description='Application to use domains from DigitalOcean account as dynamic '\
'DNS domain(s).\nThe app only supports IP4. IPv6 is planned for a later release!'\
'\nYou\'ll always find the latest version on https://gitlab.pm/rune/ddns\n\n'\
'For bugs, suggestions, pull requests visit https://gitlab.pm/rune/ddns/issues',
formatter_class=RawTextHelpFormatter,
epilog='Making Selfhosting easier...')
parser.add_argument('-a', '--api', help='Add/Change API key.\n\n',
nargs=1, metavar=('APIkey'), required=False, action="append")
parser.add_argument('-f', '--force', help='Force update of IP address for all domains.\n\n',
required=False, action="store_true")
parser.add_argument('-l', '--list', help='List subdomains for supplied domain.\n\n',
nargs=1, metavar=('domain'), required=False, action="append")
parser.add_argument('-o', '--serverdomains', help='List subdomains for supplied domain not in ddns DB.\n\n',
nargs=1, metavar=('domain'), required=False, action="append")
parser.add_argument('-d', '--domains', help='List top domains in your DigitalOcean account.\n\n',
required=False, action="store_true")
parser.add_argument('-c', '--current', help='List the current IP address for the sub-domain given\n\n',
required=False, nargs=1, action="append")
parser.add_argument('-t', '--top', help='Add a new domain from your DigitalOcean account to use as a dynamic DNS domain\n\n',
required=False, nargs=1, metavar=('domain'), action='append')
parser.add_argument('-s', '--sub', help='Add a new subdomain to your DigitalOcean account and use as dynamic DNS.\n\n\n',
required=False, nargs=1, metavar=('domain'), action='append')
parser.add_argument('-k', '--local', help='Add an existing DigitalOcean subdomain to your ddns DB and use as dynamic DNS.\n\n',
required=False, nargs=2, metavar=('domain','domainid'), action='append')
parser.add_argument('-r', '--remove', help='Remove a subdomain from your DigitalOcean account and ddns.\n\n',
required=False, nargs=1, metavar=('domain'), action='append')
parser.add_argument('-v', '--version', help='Show current version and config info\n\n',
required=False, action='store_true')
parser.add_argument('-q', '--log', help=argparse.SUPPRESS, required=False, action='store_true')
parser.add_argument('-p', '--ipserver', help='Sets or updates IP server lookup to use. Indicate 4 or 6 for IP type.\n\n',
required=False, nargs=2, metavar=('ip4.iurl.no', '4'), action="append")
parser.add_argument('-e', '--edit', help='Changes domain from active to inactive or the other way around...',
required=False, nargs=1, metavar=('test.example.com'), action="append")
args = vars(parser.parse_args())
if args['list']:
list_sub_domains(args['list'][0][0])
elif args['domains']:
show_all_top_domains()
elif args['serverdomains']:
list_do_sub_domains(args['serverdomains'][0][0])
elif args['current']:
domaininfo(args['current'][0][0])
elif args['top']:
add_domian(args['top'][0][0])
elif args['sub']:
add_subdomain(args['sub'][0][0])
elif args['version']:
show_current_info()
elif args['force']:
updateip(True)
elif args['log']:
show_log()
elif args['ipserver']:
ip_server(args['ipserver'][0][0],args['ipserver'][0][1])
elif args['api']:
api(args['api'][0][0])
elif args['remove']:
remove_subdomain(args['remove'][0][0])
elif args['edit']:
edit_subdomain(args['edit'][0][0])
elif args['local']:
local_add_subdomain(args['local'][0][0],args['local'][0][1])
else:
updateip(None)
hosts/mainframe/bin/ddns_update
Executable
+108
View File
@@ -0,0 +1,108 @@
#!/bin/zsh
#
# Publish public DNS records for the Caddy site blocks that are explicitly
# marked public, and refresh the A record for the current WAN IP.
#
# A site block is published ONLY if the line immediately above it is exactly
#
# # ddns: public
#
# Anything unmarked stays off public DNS. That default matters: this script
# used to scrape *every* site block, which is how internal-only services ended
# up with public records (and, via Certificate Transparency, public hostnames).
# Forgetting a marker now fails closed.
CADDYFILE=/home/connor/Caddyfile
NTFY_ENV=/home/connor/.config/ntfy/publish.env
NTFY_URL=https://ntfy.rcjohnstone.com/infra
notify() {
# Best-effort: never let a notification failure fail the run.
[[ -r $NTFY_ENV ]] || return 0
local user pass
user=$(sed -n 's/^NTFY_USER=//p' $NTFY_ENV)
pass=$(sed -n 's/^NTFY_PASS=//p' $NTFY_ENV)
[[ -n $user && -n $pass ]] || return 0
curl -sS -m 15 -u "$user:$pass" \
-H "Title: $1" -H "Priority: ${3:-default}" -H "Tags: ${4:-warning}" \
-d "$2" $NTFY_URL >/dev/null 2>&1
return 0
}
fail() {
print -u2 "ddns_update: $1"
notify "DDNS update failed" "$1" high rotating_light
exit 1
}
ip=$(curl -sq4 -m 20 ifconfig.me)
if [[ -z "$ip" ]]; then
fail "could not determine public IPv4, aborting"
fi
if ! out=$(ddns -p $ip 4 2>&1); then
fail "ddns -p failed:\n$out"
fi
# Emit the hostname from each site block preceded by the marker. Blank lines
# between the marker and the block are tolerated; anything else resets it.
subdomains=$(awk '
/^[[:space:]]*#[[:space:]]*ddns:[[:space:]]*public[[:space:]]*$/ { pub=1; next }
/^[[:space:]]*$/ { next }
/^[^[:space:]#].*\{[[:space:]]*$/ {
if (pub) { name=$0; sub(/[{,].*/, "", name); gsub(/[[:space:]]/, "", name); print name }
pub=0; next
}
{ pub=0 }
' $CADDYFILE \
| grep -E '(^|\.)rcjohnstone\.com$' \
| sed 's/^rcjohnstone\.com$/@.rcjohnstone.com/' \
| sort -u)
if [[ -z "$subdomains" ]]; then
fail "no '# ddns: public' markers found in $CADDYFILE -- refusing to continue"
fi
added=()
for subdomain in ${(f)subdomains}
do
# already-tracked names just print "already exists"; that is not an error
if out=$(ddns -s $subdomain 2>&1); then
print -r -- "$out" | grep -v 'already exists'
print -r -- "$out" | grep -q 'already exists' || added+=$subdomain
else
print -u2 "ddns_update: failed to add $subdomain:\n$out"
fi
done
if (( ${#added} )); then
notify "DDNS: new public records" \
"Now publicly resolvable:\n${(F)added}" default globe_with_meridians
fi
# Reconcile. The bare `ddns` below refreshes every subdomain in ddns.db, not
# just the ones we asked for -- that is how `spanish` acquired a public record
# despite having no marker. Anything tracked locally but no longer marked is
# drift, and drift here means a service is on the internet that we did not
# intend to publish. Report it loudly rather than silently republishing.
tracked=$(python3 - <<'EOF' 2>/dev/null
import sqlite3
try:
c = sqlite3.connect('/home/connor/.config/ddns/ddns.db')
print("\n".join(sorted(r[0] for r in c.execute("select name from subdomains"))))
except Exception:
pass
EOF
)
if [[ -n "$tracked" ]]; then
marked=$(print -r -- "$subdomains" | sed 's/\.rcjohnstone\.com$//' | sed 's/^@$/@/' | sort -u)
drift=$(comm -23 <(print -r -- "$tracked") <(print -r -- "$marked"))
if [[ -n "$drift" ]]; then
print -u2 "ddns_update: tracked but NOT marked public in the Caddyfile:\n$drift"
notify "DDNS drift: unmarked names still published" \
"These have DNS records but no '# ddns: public' marker:\n$drift\n\nRemove them from DigitalOcean and from ddns.db, or add the marker." \
high warning
fi
fi
ddns
hosts/mainframe/bin/ds_to_traggo
Executable
+183
View File
@@ -0,0 +1,183 @@
#!/usr/bin/python3
"""Import Dreaming Spanish progress into Traggo as tagged time spans.
ds_to_traggo [--apply] [--from YYYY-MM-DD] [--include-initial]
Dry-run unless --apply is given. Safe to re-run: spans this tool created are
marked in their note and are not recreated.
How the two DS endpoints relate (verified against live data):
dayWatchedTime is the daily total of EVERYTHING -- platform videos plus
every external entry, including 'talking'.
externalTime is the itemised list of off-platform entries.
So listening must be derived by subtraction, not by adding the two together:
mode:speaking = external 'talking'
mode:listening = dayWatchedTime - talking (platform + watching + listening)
"""
import argparse
import json
import subprocess
import sys
import urllib.request
from collections import defaultdict
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
DS_BASE = "https://app.dreaming.com/.netlify/functions"
DS_TOKEN_FILE = "/tmp/dreaming_token"
TRAGGO = "https://time.rcjohnstone.com/graphql"
TZ = ZoneInfo("America/Louisville")
MARKER = "[ds-import]"
TAGS = {"language": "#4a90d9", "mode": "#7ab317"}
def ds_get(endpoint, token):
req = urllib.request.Request(
"%s/%s" % (DS_BASE, endpoint), headers={"Authorization": "Bearer " + token})
with urllib.request.urlopen(req, timeout=60) as r:
return json.loads(r.read())
def gql(query, variables, token):
body = json.dumps({"query": query, "variables": variables}).encode()
req = urllib.request.Request(
TRAGGO, data=body,
headers={"Content-Type": "application/json",
"Authorization": "traggo " + token})
with urllib.request.urlopen(req, timeout=60) as r:
out = json.loads(r.read())
if out.get("errors"):
raise RuntimeError(json.dumps(out["errors"])[:300])
return out["data"]
def traggo_login():
pw = subprocess.run(["rbw", "get", "Traggo"],
capture_output=True, text=True).stdout.strip()
if not pw:
sys.exit("Could not read the Traggo password from Bitwarden (rbw get Traggo).")
q = ("mutation($u:String!,$p:String!){login(username:$u,pass:$p,"
"deviceName:\"ds-import\",type:NoExpiry,cookie:false){token}}")
return gql(q, {"u": "connor", "p": pw}, "")["login"]["token"]
def ensure_tags(token):
have = {t["key"] for t in gql("{tags{key}}", {}, token)["tags"]}
for key, color in TAGS.items():
if key not in have:
gql("mutation($k:String!,$c:String!){createTag(key:$k,color:$c){key}}",
{"k": key, "c": color}, token)
print(" created tag key: %s" % key)
def existing(token, first, last):
"""Dates already imported, as {(date, mode)}."""
q = ("query($f:Time!,$t:Time!,$c:InputCursor){timeSpans(fromInclusive:$f,"
"toInclusive:$t,cursor:$c){cursor{hasMore offset startId pageSize}"
"timeSpans{start note tags{key value}}}}")
seen, cursor = set(), {"offset": 0, "pageSize": 200}
while True:
page = gql(q, {"f": first, "t": last, "c": cursor}, token)["timeSpans"]
for ts in page["timeSpans"]:
if MARKER not in (ts.get("note") or ""):
continue
mode = next((t["value"] for t in ts["tags"] if t["key"] == "mode"), None)
seen.add((ts["start"][:10], mode))
c = page["cursor"]
if not c["hasMore"]:
return seen
cursor = {"offset": c["offset"], "startId": c["startId"],
"pageSize": c["pageSize"]}
def build(ds_token, start_from, include_initial):
ext = ds_get("externalTime", ds_token)["externalTimes"]
day = {d["date"]: d["timeSeconds"] for d in ds_get("dayWatchedTime", ds_token)}
talk, notes = defaultdict(int), defaultdict(list)
for e in ext:
if e["type"] == "initial" and not include_initial:
continue
if e["type"] == "talking":
talk[e["date"]] += e["timeSeconds"]
if e.get("description"):
notes[(e["date"], "speaking")].append(e["description"])
elif e.get("description"):
notes[(e["date"], "listening")].append(e["description"])
plan = []
for date in sorted(day):
if start_from and date < start_from:
continue
listening = day[date] - talk.get(date, 0)
speaking = talk.get(date, 0)
# Anchor at midnight local; speaking picks up where listening ends so
# the two never overlap in the calendar view. DS records daily totals
# only, so the clock times are synthetic either way.
cursor = datetime(int(date[:4]), int(date[5:7]), int(date[8:10]),
tzinfo=TZ)
for mode, secs in (("listening", listening), ("speaking", speaking)):
if secs <= 0:
continue
end = cursor + timedelta(seconds=secs)
desc = ", ".join(dict.fromkeys(notes.get((date, mode), [])))
plan.append({
"date": date, "mode": mode, "seconds": secs,
"start": cursor.isoformat(), "end": end.isoformat(),
"note": ("%s %s" % (MARKER, desc)).strip(),
})
cursor = end
return plan
def main():
p = argparse.ArgumentParser()
p.add_argument("--apply", action="store_true")
p.add_argument("--from", dest="start_from", metavar="YYYY-MM-DD")
p.add_argument("--include-initial", action="store_true",
help="also import the 50h 'time prior to Dreaming Spanish' entry")
a = p.parse_args()
ds_token = open(DS_TOKEN_FILE).read().strip()
plan = build(ds_token, a.start_from, a.include_initial)
if not plan:
print("Nothing to import.")
return
token = traggo_login()
ensure_tags(token)
already = existing(token, plan[0]["start"], plan[-1]["end"])
todo = [s for s in plan if (s["date"], s["mode"]) not in already]
hrs = lambda rows, m: sum(r["seconds"] for r in rows if r["mode"] == m) / 3600
print(" planned : %d spans (%.1fh listening, %.1fh speaking) over %s..%s"
% (len(plan), hrs(plan, "listening"), hrs(plan, "speaking"),
plan[0]["date"], plan[-1]["date"]))
print(" already : %d spans present from a previous run" % (len(plan) - len(todo)))
print(" to create: %d spans (%.1fh listening, %.1fh speaking)"
% (len(todo), hrs(todo, "listening"), hrs(todo, "speaking")))
if not a.apply:
print("\n first 5:")
for s in todo[:5]:
print(" %s %-9s %5.0fm %s" % (s["date"], s["mode"],
s["seconds"] / 60, s["note"][:48]))
print("\n Dry run. Re-run with --apply to write these to Traggo.")
return
q = ("mutation($s:Time!,$e:Time!,$t:[InputTimeSpanTag!],$n:String!){"
"createTimeSpan(start:$s,end:$e,tags:$t,note:$n){id}}")
made = 0
for s in todo:
tags = [{"key": "language", "value": "spanish"},
{"key": "mode", "value": s["mode"]}]
try:
gql(q, {"s": s["start"], "e": s["end"], "t": tags, "n": s["note"]}, token)
made += 1
except Exception as err:
print(" FAILED %s %s: %s" % (s["date"], s["mode"], err))
print(" created %d/%d spans" % (made, len(todo)))
if __name__ == "__main__":
main()
hosts/mainframe/bin/inbox_tidy
Executable
+443
View File
@@ -0,0 +1,443 @@
#!/usr/bin/python3
"""Archive stale unread mail out of the Proton inbox. Never deletes.
The inbox is meant to hold new mail plus read mail kept on purpose. Unread mail
accumulates faster than it gets triaged, so anything still unread after a few
days is moved to Archive.
Deliberately has no LLM in it. Of ~1200 inbox messages sampled, 81% carry a
machine-readable bulk marker, and the ~19% residual is transactional rather
than personal. Since transactional mail is archived on the same rule, every
decision reduces to a flag test, a set membership test or a string match --
deterministic, explainable in the log, and not improved by a model.
Safety, in order of importance:
* nothing is ever deleted; the only operation is IMAP MOVE to Archive
* BODY.PEEK everywhere, so nothing is ever marked read as a side effect
* INBOX is the only mailbox ever opened read-write
* every archived message is recorded with its Message-ID and the rule that
fired, and any run can be undone
* a message with no Message-ID is left alone, because it could not be undone
* two interlocks abort the run if the mailbox does not look like itself
"""
import argparse
import datetime as dt
import email
import email.policy
import email.utils
import json
import os
import re
import sys
import time
import tomllib
# Shared with rent_utilities. sys.path rather than PYTHONPATH so this works
# when run by hand as well as from the unit.
sys.path.insert(0, os.path.expanduser("~/.local/lib/pymail"))
from protonimap import ( # noqa: E402
Fatal, imap_connect, imap_date, notify_failure, push, _quote)
CONFIG = os.path.expanduser("~/.config/inbox-tidy/tidy.toml")
LOG = os.path.expanduser("~/.local/share/inbox-tidy/archived.jsonl")
HEADERS = ("FROM SUBJECT DATE MESSAGE-ID LIST-UNSUBSCRIBE LIST-ID PRECEDENCE "
"AUTO-SUBMITTED FEEDBACK-ID X-CAMPAIGNID")
BULK_HEADERS = ("List-Unsubscribe", "List-Id", "Feedback-ID", "X-CampaignID")
# Skip reasons, in evaluation order. Also the keys in the run summary.
KEPT_FLAGGED = "flagged-or-answered"
KEPT_SENDER = "protected-sender"
KEPT_CORRESPONDENT = "known-correspondent"
KEPT_SUBJECT = "protected-subject"
KEPT_NO_MSGID = "no-message-id"
ARCHIVE = "archive"
# --------------------------------------------------------------------------
# helpers
# --------------------------------------------------------------------------
def load_config(path=None):
path = path or CONFIG
try:
with open(path, "rb") as fh:
return tomllib.load(fh)
except (OSError, tomllib.TOMLDecodeError) as exc:
raise Fatal("Cannot read %s: %s" % (path, exc))
def addr_of(header_value):
"""Bare lowercase address out of a From/To header, or ''."""
if not header_value:
return ""
pairs = email.utils.getaddresses([str(header_value)])
return pairs[0][1].lower().strip() if pairs and pairs[0][1] else ""
def domain_of(addr):
return addr.rsplit("@", 1)[-1] if "@" in addr else ""
def sender_protected(addr, entries):
"""An entry containing @ matches the address; otherwise the domain.
Domain entries match subdomains too, so `chase.com` also covers
`fraudalert.chase.com` -- which is the case that actually matters.
"""
dom = domain_of(addr)
for e in entries:
e = e.lower().strip()
if not e:
continue
if "@" in e:
if addr == e:
return True
elif dom == e or dom.endswith("." + e):
return True
return False
def fetch_headers(conn, ids, extra="FLAGS"):
"""[(uid, flags, email.Message)] for a list of sequence ids, batched.
BODY.PEEK, never BODY: fetching a body with BODY would set \\Seen on mail
this job exists to leave unread.
"""
out = []
for i in range(0, len(ids), 200):
chunk = b",".join(ids[i:i + 200]).decode()
typ, data = conn.fetch(
chunk, "(UID %s BODY.PEEK[HEADER.FIELDS (%s)])" % (extra, HEADERS))
if typ != "OK":
continue
for item in data:
if not isinstance(item, tuple):
continue
meta = item[0].decode(errors="replace")
m_uid = re.search(r"UID (\d+)", meta)
m_fl = re.search(r"FLAGS \(([^)]*)\)", meta)
if not m_uid:
continue
msg = email.message_from_bytes(item[1], policy=email.policy.default)
out.append((m_uid.group(1), (m_fl.group(1) if m_fl else ""), msg))
return out
def correspondents(conn, cfg):
"""Addresses this account has actually written to, from the Sent folder.
A behavioural allowlist: someone you have emailed is someone whose mail
should not be swept up. Cheap to derive and needs no maintenance.
"""
box = cfg["imap"].get("sent_mailbox", "Sent")
typ, _ = conn.select(_quote(box), readonly=True)
if typ != "OK":
raise Fatal("Cannot open the Sent mailbox %r to build the "
"correspondent allowlist." % box)
typ, data = conn.search(None, "ALL")
ids = data[0].split() if typ == "OK" else []
ids = ids[-cfg["imap"].get("sent_scan_limit", 3000):]
found = set()
for i in range(0, len(ids), 200):
chunk = b",".join(ids[i:i + 200]).decode()
typ, data = conn.fetch(
chunk, "(BODY.PEEK[HEADER.FIELDS (TO CC BCC)])")
if typ != "OK":
continue
for item in data:
if not isinstance(item, tuple):
continue
h = email.message_from_bytes(item[1], policy=email.policy.default)
for field in ("To", "Cc", "Bcc"):
for _, a in email.utils.getaddresses(h.get_all(field, [])):
if a and "@" in a:
found.add(a.lower().strip())
return found
def decide(msg, flags, cfg, corr):
"""-> (outcome, detail). Guards first; anything uncertain is kept."""
if "\\Flagged" in flags or "\\Answered" in flags:
return KEPT_FLAGGED, flags.strip()
frm = addr_of(msg.get("From"))
if sender_protected(frm, cfg.get("protect_senders", [])):
return KEPT_SENDER, frm
if frm and frm in corr:
return KEPT_CORRESPONDENT, frm
subject = str(msg.get("Subject", ""))
for pat in cfg.get("protect_subjects", []):
if re.search(pat, subject, re.I):
return KEPT_SUBJECT, pat
# Undo works by Message-ID. Without one the move could not be reversed,
# so it is not made -- a message stuck in the inbox is a far smaller
# problem than one that cannot be brought back.
if not str(msg.get("Message-ID", "")).strip():
return KEPT_NO_MSGID, ""
return ARCHIVE, ""
def bulk_markers(msg):
return [h for h in BULK_HEADERS if msg.get(h)] + (
["Precedence"] if (msg.get("Precedence") or "").lower().strip()
in ("bulk", "list", "junk") else [])
def log_records(records):
with open(LOG, "a") as fh:
for r in records:
fh.write(json.dumps(r, sort_keys=True) + "\n")
def read_log():
try:
with open(LOG) as fh:
return [json.loads(l) for l in fh if l.strip()]
except FileNotFoundError:
return []
# --------------------------------------------------------------------------
# modes
# --------------------------------------------------------------------------
def gather(conn, cfg, args):
"""-> (decisions, corr) where decisions is [(uid, msg, outcome, detail)]."""
corr = correspondents(conn, cfg)
minc = cfg.get("min_correspondents", 25)
if len(corr) < minc and not args.check:
raise Fatal(
"The correspondent allowlist has only %d address(es); %d are "
"required.\n\n"
"That guard is derived from the Sent mailbox, so a nearly empty "
"Sent means the guard is silently doing nothing -- mail from "
"people you actually write to would be archived like any "
"circular. Refusing to run.\n\n"
"If Bridge is still doing its first sync, wait for it to finish "
"and try again:\n"
" sudo podman logs --tail 5 connor_protonmail-bridge_1\n"
"Lower `min_correspondents` in %s only if this account genuinely "
"sends very little mail." % (len(corr), minc, CONFIG))
conn.select("INBOX", readonly=True)
cutoff = dt.date.today() - dt.timedelta(days=cfg.get("age_days", 3))
typ, data = conn.search(None, '(UNSEEN BEFORE "%s")' % imap_date(cutoff))
if typ != "OK":
raise Fatal("IMAP search of INBOX failed: %s" % typ)
ids = data[0].split()
cap = cfg.get("expected_unread_max", 900)
if len(ids) > cap and not args.check:
raise Fatal(
"%d messages in INBOX are unread and older than %d days, which is "
"above the sanity limit of %d.\n\n"
"This limit exists because Bridge's \\Seen flags did not initially "
"match the Proton UI -- it reported 1249 unread where the UI "
"showed 646. Archiving on wrong flags would move READ mail, which "
"is exactly the mail the inbox is meant to keep.\n\n"
"Compare `inbox_tidy --check` against the unread count in the "
"Proton web UI. If they agree, this really is the backlog and "
"`expected_unread_max` in %s should be raised. If they disagree, "
"do not run this." % (len(ids), cfg.get("age_days", 3), cap, CONFIG))
decisions = []
for uid, flags, msg in fetch_headers(conn, ids):
outcome, detail = decide(msg, flags, cfg, corr)
decisions.append((uid, msg, outcome, detail))
return decisions, corr
def do_check(conn, cfg):
print("Mailbox state (compare INBOX unseen against the Proton web UI):\n")
for box in ("INBOX", "Archive", cfg["imap"].get("sent_mailbox", "Sent")):
typ, st = conn.status(_quote(box), "(MESSAGES UNSEEN)")
print(" %-10s %s" % (box, st[0].decode() if typ == "OK" else "?"))
corr = correspondents(conn, cfg)
conn.select("INBOX", readonly=True)
cutoff = dt.date.today() - dt.timedelta(days=cfg.get("age_days", 3))
typ, data = conn.search(None, '(UNSEEN BEFORE "%s")' % imap_date(cutoff))
n = len(data[0].split()) if typ == "OK" else 0
print("\n correspondents from Sent : %d (minimum %d)"
% (len(corr), cfg.get("min_correspondents", 25)))
print(" candidates (unread >%dd) : %d (sanity limit %d)"
% (cfg.get("age_days", 3), n, cfg.get("expected_unread_max", 900)))
print(" enabled : %s" % cfg.get("enabled", False))
print("\nProceed only if INBOX unseen matches the web UI and the "
"correspondent count looks real.")
return 0
def do_run(conn, cfg, args):
decisions, corr = gather(conn, cfg, args)
to_move = [(u, m) for u, m, o, _ in decisions if o == ARCHIVE]
limit = args.max if args.max is not None else cfg.get("max_per_run", 250)
batch, held = to_move[:limit], to_move[limit:]
counts = {}
for _, _, outcome, _ in decisions:
counts[outcome] = counts.get(outcome, 0) + 1
if args.dry_run:
for uid, msg, outcome, detail in decisions:
print("%-20s %-34s %s" % (
outcome, addr_of(msg.get("From"))[:34],
str(msg.get("Subject", ""))[:60]) +
((" [%s]" % detail) if detail and outcome != ARCHIVE else ""))
print("\n%s" % summarise(counts, len(batch), len(held), None))
return 0
if not cfg.get("enabled", False):
raise Fatal(
"`enabled` is false in %s.\n\nThe job is installed but not armed. "
"Review `inbox_tidy --check` and `--dry-run` first, then set "
"enabled = true." % CONFIG)
run_id = time.strftime("%Y%m%dT%H%M%S")
moved = []
if batch:
conn.select("INBOX", readonly=False) # the ONLY read-write select
for i in range(0, len(batch), 100):
part = batch[i:i + 100]
typ, _ = conn.uid("MOVE", ",".join(u for u, _ in part), '"Archive"')
if typ != "OK":
raise Fatal(
"IMAP MOVE failed after %d message(s) of this run.\n\n"
"Anything already moved is in Archive and is recorded in "
"%s under run %s, so `inbox_tidy --undo %s` will bring it "
"back." % (len(moved), LOG, run_id, run_id))
for uid, msg in part:
moved.append({
"run": run_id,
"at": dt.datetime.now(dt.timezone.utc).isoformat(),
"action": "archive",
"uid": uid,
"message_id": str(msg.get("Message-ID", "")).strip(),
"from": addr_of(msg.get("From")),
"subject": str(msg.get("Subject", ""))[:200],
"date": str(msg.get("Date", "")),
"bulk": bulk_markers(msg),
"src": "INBOX", "dst": "Archive",
})
log_records(moved[-len(part):])
body = summarise(counts, len(moved), len(held), run_id)
push(cfg, "Inbox tidy: %d archived" % len(moved), body, "broom")
print(body)
return 0
def summarise(counts, n_moved, n_held, run_id):
lines = ["**%d archived**" % n_moved]
if n_held:
lines.append("%d over the per-run cap, left for the next run." % n_held)
lines.append("")
lines.append("Kept:")
for k in (KEPT_FLAGGED, KEPT_CORRESPONDENT, KEPT_SENDER, KEPT_SUBJECT,
KEPT_NO_MSGID):
if counts.get(k):
lines.append(" %-22s %d" % (k, counts[k]))
if not any(counts.get(k) for k in (KEPT_FLAGGED, KEPT_CORRESPONDENT,
KEPT_SENDER, KEPT_SUBJECT,
KEPT_NO_MSGID)):
lines.append(" (nothing was held back by a guard)")
if run_id:
lines += ["", "Undo: `inbox_tidy --undo %s`" % run_id]
return "\n".join(lines)
def do_undo(conn, cfg, run_id):
recs = [r for r in read_log()
if r.get("action") == "archive" and r.get("run") == run_id]
if not recs:
raise Fatal("No archive records for run %r in %s." % (run_id, LOG))
conn.select('"Archive"', readonly=False)
restored, missing = [], []
for r in recs:
mid = r.get("message_id", "")
if not mid:
missing.append(r)
continue
typ, data = conn.search(None, '(HEADER MESSAGE-ID "%s")' % mid)
ids = data[0].split() if typ == "OK" else []
if not ids:
missing.append(r)
continue
typ, d2 = conn.fetch(b",".join(ids).decode(), "(UID)")
uids = [m.group(1) for m in
(re.search(r"UID (\d+)", x.decode(errors="replace"))
for x in d2 if isinstance(x, bytes)) if m]
if not uids:
missing.append(r)
continue
typ, _ = conn.uid("MOVE", ",".join(uids), "INBOX")
(restored if typ == "OK" else missing).append(r)
log_records([{
"run": run_id, "at": dt.datetime.now(dt.timezone.utc).isoformat(),
"action": "undo", "restored": len(restored), "missing": len(missing),
}])
body = ("**Undo of run %s**\n\n%d restored to INBOX.\n%d not found in "
"Archive (moved or deleted by hand since)."
% (run_id, len(restored), len(missing)))
print(body)
push(cfg, "Inbox tidy: undo %s" % run_id, body, "leftwards_arrow_with_hook")
return 0 if not missing else 1
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--config", metavar="PATH",
help="use an alternative config (for testing a rule change "
"without touching the live one)")
ap.add_argument("--check", action="store_true",
help="report mailbox state and the guard inputs; change nothing")
ap.add_argument("--dry-run", action="store_true",
help="print the decision for every candidate; move nothing")
ap.add_argument("--max", type=int, help="override the per-run cap")
ap.add_argument("--undo", metavar="RUN_ID", help="restore a run from Archive")
ap.add_argument("--undo-last", action="store_true",
help="restore the most recent archive run")
args = ap.parse_args()
try:
cfg = load_config(args.config)
except Fatal as exc:
print("inbox_tidy: %s" % exc, file=sys.stderr)
return 2
conn = None
try:
conn = imap_connect(cfg["imap"])
if args.undo or args.undo_last:
run_id = args.undo
if args.undo_last:
runs = [r["run"] for r in read_log() if r.get("action") == "archive"]
if not runs:
raise Fatal("Nothing in %s to undo." % LOG)
run_id = runs[-1]
return do_undo(conn, cfg, run_id)
if args.check:
return do_check(conn, cfg)
return do_run(conn, cfg, args)
except Fatal as exc:
body = "**Inbox tidy could not run.**\n\n%s" % exc
if args.dry_run or args.check:
print(body, file=sys.stderr)
else:
notify_failure(cfg, "Inbox tidy: FAILED", body)
return 1
finally:
if conn is not None:
try:
conn.logout()
except Exception:
pass
if __name__ == "__main__":
sys.exit(main())
hosts/mainframe/bin/movie_recs
Executable
+463
View File
@@ -0,0 +1,463 @@
#!/usr/bin/python3
"""Recommend movies based on the whole Radarr library, via the local LLM stack.
movie_recs [-n COUNT] [-m MODEL] [--json] [--prompt-only]
Talks to LiteLLM (OpenAI-compatible) rather than Ollama, which was removed
2026-08-28. Ollama and llama-server both wanted the whole 3060, and with
gpt-oss-20b resident at ~11.3 GB of 11.9 GB there was under 1 GB left, so
every Ollama request OOM'd. llama-swap now arbitrates a single GPU owner.
The model is the same weights as before (gemma3 12B instruct, Q4_K_M), just
served by llama.cpp instead, so recommendations keep their prior character.
Reads every movie in Radarr, asks the model for titles that are NOT already
in the library, and filters the answer against the library again (models
reliably recommend things you already own).
"""
import argparse
import json
import os
import re
from collections import Counter, defaultdict
import sys
import unicodedata
import urllib.error
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
RADARR_CONFIG = "/home/connor/config/radarr/config.xml"
# Router DNS override points this at the LAN, so it never leaves the network.
RADARR_URL = "https://radarr.rcjohnstone.com"
# LiteLLM's loopback publish -- no reason to make it a network round trip.
# (Caddy at llm.rcjohnstone.com is the LAN/OpenVPN path; this runs on the host.)
LITELLM_URL = "http://127.0.0.1:4000"
# A LiteLLM virtual key scoped to just this model, NOT the master key: this
# runs unattended from a timer, and a scoped key also gives the spend log a
# per-consumer attribution instead of lumping every caller under "master".
# 0600 file, same pattern as movie_recs_notify's ntfy credential.
LITELLM_ENV = os.path.expanduser("~/.config/litellm/movie-recs.env")
SCHEMA = {
"type": "object",
"properties": {
"recommendations": {
"type": "array",
"items": {
"type": "object",
"properties": {
"title": {"type": "string"},
"year": {"type": "integer"},
"anchors": {
"type": "array",
"items": {"type": "string"},
"minItems": 3,
"maxItems": 5,
},
"pattern": {"type": "string"},
"reason": {"type": "string"},
},
"required": ["title", "year", "anchors", "pattern", "reason"],
},
}
},
"required": ["recommendations"],
}
def norm(t):
"""Loose title key: fold accents, drop articles/punctuation/case."""
t = unicodedata.normalize("NFKD", t)
t = "".join(c for c in t if not unicodedata.combining(c)).lower()
t = re.sub(r"[^a-z0-9 ]+", "", t)
t = re.sub(r"^(the|a|an) ", "", t).strip()
return re.sub(r"\s+", " ", t)
def get_json(url, data=None, timeout=600, headers=None):
req = urllib.request.Request(
url,
data=json.dumps(data).encode() if data is not None else None,
headers={"Content-Type": "application/json", **(headers or {})},
)
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read().decode())
def litellm_key():
"""Read the virtual key, env first so a manual run can override."""
key = os.environ.get("LITELLM_KEY")
if key:
return key
try:
with open(LITELLM_ENV) as fh:
for line in fh:
line = line.strip()
if line.startswith("#") or "=" not in line:
continue
k, v = line.split("=", 1)
if k.strip() == "LITELLM_KEY":
return v.strip()
except OSError:
pass
sys.exit("No LiteLLM key: set LITELLM_KEY or write it to %s" % LITELLM_ENV)
def api_key():
return ET.parse(RADARR_CONFIG).getroot().findtext("ApiKey")
def library(key):
movies = get_json("%s/api/v3/movie?apikey=%s" % (RADARR_URL, key), timeout=60)
out = []
for m in movies:
out.append({
"title": m.get("title", "?"),
"year": m.get("year"),
"genres": m.get("genres") or [],
})
return sorted(out, key=lambda x: (x["year"] or 0))
def verify(key, title, year):
"""Resolve a suggestion against TMDB via Radarr's own lookup.
Models at this size invent plausible-sounding films, so a title the
model asserts is only accepted when TMDB agrees on both the name and
(within a year) the release date. Returns the real record or None.
"""
# Models often glue the year onto the title ("Gone Girl (2014)"), which
# would never compare equal to TMDB's bare title. Split it back out.
m = re.search(r"\((\d{4})\)\s*$", title)
if m:
year = year or int(m.group(1))
title = title[:m.start()].strip()
url = "%s/api/v3/movie/lookup?apikey=%s&term=%s" % (
RADARR_URL, key, urllib.parse.quote(title))
try:
hits = get_json(url, timeout=30)
except Exception:
return None
want = norm(title)
matches = [h for h in hits[:10] if norm(h.get("title", "")) == want]
if not matches and year:
# Films are often listed under a longer official title ("The French
# Dispatch" vs "The French Dispatch of the Liberty, Kansas Evening
# Sun"). Accept a prefix, but only with the year agreeing, so this
# stays tight enough to keep rejecting invented titles.
matches = [h for h in hits[:10]
if norm(h.get("title", "")).startswith(want + " ")
and h.get("year") and abs(int(h["year"]) - int(year)) <= 1]
if not matches:
return None
# The title existing is the real signal; models get years wrong on films
# that are perfectly real. Use the year only to disambiguate remakes, and
# let TMDB's value win.
if year:
matches.sort(key=lambda h: abs((h.get("year") or 0) - int(year)))
h = matches[0]
imdb = (h.get("ratings") or {}).get("imdb") or {}
return {
"title": h.get("title"),
"year": h.get("year"),
"tmdbId": h.get("tmdbId"),
"overview": (h.get("overview") or "").strip(),
"genres": h.get("genres") or [],
"runtime": h.get("runtime"),
"votes": imdb.get("votes") or 0,
"score": imdb.get("value") or 0,
}
def decade_mix(lib):
c = Counter((m["year"] // 10) * 10 for m in lib if m.get("year"))
total = sum(c.values()) or 1
return c, total
def decade_targets(lib, n):
"""How many of n picks each decade should get, mirroring the library."""
c, total = decade_mix(lib)
raw = {d: n * v / total for d, v in c.items()}
base = {d: int(v) for d, v in raw.items()}
for d, _ in sorted(raw.items(), key=lambda kv: -(kv[1] - int(kv[1]))):
if sum(base.values()) >= n:
break
base[d] += 1
return base
def build_prompt(lib, n):
# Newest first: the collection is weighted toward recent films, and a
# list that opens in the 1940s drags recommendations into the past.
ordered = sorted(lib, key=lambda m: -(m["year"] or 0))
lines = ["%s (%s) [%s]" % (m["title"], m["year"], ", ".join(m["genres"][:3]))
for m in ordered]
c, total = decade_mix(lib)
hist = ", ".join("%ss %d%%" % (d, round(100 * c[d] / total))
for d in sorted(c, reverse=True))
# Spell out the per-decade counts. Given only a percentage breakdown the
# model swings to whichever end the wording emphasizes.
tgt = decade_targets(lib, n)
quota = "; ".join("%d from the %ss" % (v, d)
for d, v in sorted(tgt.items(), reverse=True) if v)
return (
"Here is my complete personal film collection (%d titles):\n\n%s\n\n"
"By decade my collection breaks down as: %s.\n\n"
"Recommend exactly %d films that are NOT in that list.\n"
"Match my collection's era spread. Give me approximately: %s.\n"
"Count as you go and respect those per-decade numbers -- a list that is "
"mostly old films is wrong, and so is one that is all recent films.\n\n"
"Base them on the collection as a whole: the recurring directors, eras, "
"genres, tones and preoccupations it reveals. Infer taste from what is "
"there.\n"
"I am trying to grow this collection, so a film being popular is fine -- "
"widely loved films are widely loved for a reason. Aim for a spread: "
"roughly half well-known films most people would recognize, and half "
"less obvious picks. What I want to avoid is a list chosen purely by "
"popularity that ignores what my collection actually says about my taste.\n"
"Work from CLUSTERS, not single films. For each recommendation, first "
"find at least three films in my list that share something real -- a "
"director, a mood, a era, a recurring theme, a kind of storytelling -- "
"then recommend a film that belongs with that group. A pick that merely "
"resembles one film I own is not useful; a pick that sits in the middle "
"of several is.\n"
"Rules:\n"
"- Never recommend a film already in the list. Check carefully.\n"
"- Real, released films only, with correct release years.\n"
"- 'title' must be the bare title only. Put the year in 'year', never in 'title'.\n"
"- 'anchors': 3 to 5 titles copied EXACTLY from my list that form the cluster.\n"
"- 'pattern': what those anchor films share, in one phrase.\n"
"- 'reason': one sentence on why the recommendation belongs with them.\n"
"- Use a different cluster for each recommendation.\n"
% (len(lines), "\n".join(lines), hist, n, quota)
)
def ask(model, prompt, n, timeout=1800):
# OpenAI-shaped, because the backend is LiteLLM -> llama-swap ->
# llama-server now. Three things moved when Ollama went away:
#
# /api/chat -> /v1/chat/completions
# "format": <schema> -> "response_format": {"type": "json_schema", ...}
# llama.cpp compiles the schema to a GBNF grammar
# and constrains sampling, same guarantee Ollama's
# "format" gave.
# "options".num_ctx -> gone. Context is a server-side flag now (-c
# 32768 in config/llama-swap/config.yaml); a
# client cannot resize it per request.
#
# max_tokens is explicit because the OpenAI schema defaults it to a finite
# value on some backends, and 60 recommendations of five fields each is a
# few thousand tokens of JSON. Truncated JSON fails the parse below rather
# than silently returning a short list.
body = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"stream": False,
"response_format": {
"type": "json_schema",
"json_schema": {"name": "movie_recommendations", "schema": SCHEMA},
},
"temperature": 0.85,
"max_tokens": 16384,
}
try:
resp = get_json(LITELLM_URL + "/v1/chat/completions", body,
timeout=timeout,
headers={"Authorization": "Bearer " + litellm_key()})
except urllib.error.HTTPError as e:
sys.exit("LiteLLM error %s: %s" % (e.code, e.read().decode()[:400]))
choice = (resp.get("choices") or [{}])[0]
content = (choice.get("message") or {}).get("content") or ""
# A truncated generation yields JSON that will not parse, and the parse
# error alone does not say why. Name the real cause.
if choice.get("finish_reason") == "length":
sys.exit("Model hit max_tokens before closing the JSON -- lower "
"--oversample or raise max_tokens in ask().")
content = re.sub(r"<think>.*?</think>", "", content, flags=re.S).strip()
if not content:
sys.exit("Model returned nothing (thinking-only response).")
try:
return json.loads(content).get("recommendations", [])
except json.JSONDecodeError:
m = re.search(r"\{.*\}", content, re.S)
if not m:
sys.exit("Could not parse model output:\n" + content[:600])
return json.loads(m.group()).get("recommendations", [])
def tier(votes):
"""Rough 'how widely seen is this' bucket, by IMDb vote count."""
if votes >= 200_000:
return "widely seen"
if votes >= 50_000:
return "known"
return "deeper cut"
def select(cands, n, balance, targets):
"""Choose n recommendations matching the library's era distribution.
Models drift toward the pre-1980 canon regardless of what the collection
looks like, so era is enforced rather than requested: each decade gets a
quota proportional to its share of the library. Within a decade, picks
alternate between better- and lesser-known films so the result is not all
blockbusters. Short buckets backfill from the most recent decades, since
that is where the collection's mass sits.
"""
if not balance:
return cands[:n]
buckets = defaultdict(list)
for c in cands:
buckets[((c["tmdb"]["year"] or 0) // 10) * 10].append(c)
for d, items in buckets.items():
ranked = sorted(items, key=lambda c: -c["tmdb"]["votes"])
half = max(len(ranked) // 2, 1)
hi, lo = ranked[:half], ranked[half:]
mixed = []
while hi or lo:
if hi:
mixed.append(hi.pop(0))
if lo:
mixed.append(lo.pop(0))
buckets[d] = mixed
out = []
for d in sorted(targets, reverse=True):
q = targets[d]
out += buckets[d][:q]
buckets[d] = buckets[d][q:]
if len(out) < n:
# Round-robin the shortfall across decades rather than draining the
# newest bucket, which would just re-create the skew the quota fixes.
order = sorted(buckets, key=lambda d: (-targets.get(d, 0), -d))
while len(out) < n and any(buckets[d] for d in order):
for d in order:
if len(out) >= n:
break
if buckets[d]:
out.append(buckets[d].pop(0))
return sorted(out[:n], key=lambda c: -(c["tmdb"]["year"] or 0))
def main():
p = argparse.ArgumentParser()
p.add_argument("-n", "--count", type=int, default=15)
p.add_argument("-m", "--model", default="gemma3-12b",
help="LiteLLM model name (see /v1/models), not an Ollama tag")
p.add_argument("--json", action="store_true")
p.add_argument("--prompt-only", action="store_true")
p.add_argument("--min-votes", type=int, default=5000,
help="drop films below this many IMDb votes (0 disables)")
p.add_argument("--no-balance", dest="balance", action="store_false",
help="skip the popularity spread and take the model's order")
p.add_argument("--timeout", type=int, default=1800,
help="seconds to wait on LiteLLM; a cold model load into "
"llama-swap costs ~30 s on top of generation")
p.add_argument("--oversample", type=int, default=4,
help="ask for COUNT*N candidates; higher fills thin decades")
a = p.parse_args()
key = api_key()
lib = library(key)
# Over-generate: a chunk of any small model's output is fabricated or
# already owned, and both get filtered below. Extra headroom also helps
# fill the thinly-populated decades.
n_candidates = a.count * a.oversample
prompt = build_prompt(lib, n_candidates)
if a.prompt_only:
print(prompt)
return
print("library: %d films | model: %s | asking for %d, keeping %d..."
% (len(lib), a.model, n_candidates, a.count), file=sys.stderr)
recs = ask(a.model, prompt, n_candidates, a.timeout)
owned = {norm(m["title"]) for m in lib}
by_norm = {norm(m["title"]): m["title"] for m in lib}
cands, invented, already, obscure, thin = [], [], [], [], []
seen = set()
for r in recs:
title = r.get("title", "")
if norm(title) in owned:
already.append(title)
continue
real = verify(key, title, r.get("year"))
if not real:
invented.append("%s (%s)" % (title, r.get("year")))
continue
if norm(real["title"]) in owned:
already.append(real["title"])
continue
if real["tmdbId"] in seen: # models repeat themselves
continue
# Anchors are only meaningful if they name films actually in the
# library -- otherwise the "cluster" is invented.
anchors = []
for anc in r.get("anchors", []):
hit = by_norm.get(norm(re.sub(r"\s*\(\d{4}\)\s*$", "", anc)))
if hit and hit not in anchors:
anchors.append(hit)
if len(anchors) < 2:
thin.append("%s [%d/%d anchors real]"
% (real["title"], len(anchors), len(r.get("anchors", []))))
continue
if a.min_votes and real["votes"] and real["votes"] < a.min_votes:
obscure.append("%s (%s, %dk votes)"
% (real["title"], real["year"], real["votes"] // 1000))
continue
seen.add(real["tmdbId"])
cands.append({**r, "tmdb": real, "anchors": anchors})
kept = select(cands, a.count, a.balance, decade_targets(lib, a.count))
if a.json:
print(json.dumps({"recommendations": kept,
"rejected_not_in_tmdb": invented,
"rejected_already_owned": already,
"rejected_below_min_votes": obscure,
"rejected_invented_cluster": thin},
indent=2, ensure_ascii=False))
return
for r in kept:
t = r["tmdb"]
print("\n\033[1m%s\033[0m (%s) \033[2m%s/10, %sk IMDb votes - %s\033[0m"
% (t["title"], t["year"], t["score"], t["votes"] // 1000, tier(t["votes"])))
print(" %s" % r.get("reason", "").strip())
print(" \033[2mcluster (%s): %s\033[0m"
% (r.get("pattern", "").strip(), ", ".join(r["anchors"])))
spread = Counter(tier(r["tmdb"]["votes"]) for r in kept)
got = Counter(((r["tmdb"]["year"] or 0) // 10) * 10 for r in kept)
want = decade_targets(lib, a.count)
lib_years = sorted(m["year"] for m in lib if m.get("year"))
rec_years = sorted(r["tmdb"]["year"] for r in kept if r["tmdb"]["year"])
print("\npopularity: %s"
% (", ".join("%s %d" % (k, v) for k, v in spread.items()) or "-"),
file=sys.stderr)
print("eras (got/target): %s"
% ", ".join("%ss %d/%d" % (d, got.get(d, 0), want.get(d, 0))
for d in sorted(set(got) | set(want), reverse=True)),
file=sys.stderr)
if rec_years:
print("median year: recommendations %d vs library %d"
% (rec_years[len(rec_years) // 2], lib_years[len(lib_years) // 2]),
file=sys.stderr)
print("%d shown | %d not real (%s) | %d owned | %d too obscure (%s) | %d invented cluster (%s)"
% (len(kept),
len(invented), ", ".join(invented[:3]) or "-",
len(already),
len(obscure), ", ".join(obscure[:3]) or "-",
len(thin), ", ".join(thin[:3]) or "-"),
file=sys.stderr)
if __name__ == "__main__":
main()
hosts/mainframe/bin/movie_recs_notify
Executable
+109
View File
@@ -0,0 +1,109 @@
#!/usr/bin/python3
"""Run movie_recs and push the result to ntfy as a markdown notification.
Intended for the movie-recs.timer systemd unit. Failures are pushed too --
a recommendation job that silently stops producing is worse than a noisy one.
"""
import base64
import json
import os
import subprocess
import sys
import urllib.request
NTFY = "https://ntfy.rcjohnstone.com/movies"
# ntfy is auth-default-access: deny-all, so publishing needs the `bot`
# credential. Mirrored from the rbw entry `ntfy-bot`; rbw is the source of
# truth. Kept in a 0600 file because this runs unattended from a timer.
NTFY_ENV = os.path.expanduser("~/.config/ntfy/publish.env")
RECS = "/home/connor/.local/bin/movie_recs"
COUNT = "10"
def _auth_header():
user, pw = os.environ.get("NTFY_USER"), os.environ.get("NTFY_PASS")
if not (user and pw):
try:
with open(NTFY_ENV) as fh:
vals = dict(
line.strip().split("=", 1)
for line in fh if "=" in line and not line.startswith("#"))
user, pw = vals.get("NTFY_USER"), vals.get("NTFY_PASS")
except OSError:
return None
if not (user and pw):
return None
token = base64.b64encode(f"{user}:{pw}".encode()).decode()
return f"Basic {token}"
def push(title, body, tags, priority="default", markdown=True):
headers = {
"Title": title,
"Tags": tags,
"Priority": priority,
"Content-Type": "text/plain; charset=utf-8",
}
auth = _auth_header()
if auth:
headers["Authorization"] = auth
if markdown:
headers["Markdown"] = "yes"
req = urllib.request.Request(
NTFY, data=body.encode("utf-8"), headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=30) as r:
return r.status
def main():
try:
proc = subprocess.run(
[RECS, "-n", COUNT, "--json"],
capture_output=True, text=True, timeout=2400)
except subprocess.TimeoutExpired:
push("Movie recs failed", "movie_recs timed out after 40 minutes.",
"warning", "high", markdown=False)
return 1
if proc.returncode != 0:
tail = (proc.stderr or "no stderr").strip().splitlines()[-6:]
push("Movie recs failed",
"exit %d\n\n%s" % (proc.returncode, "\n".join(tail)),
"warning", "high", markdown=False)
return 1
try:
data = json.loads(proc.stdout)
except json.JSONDecodeError:
push("Movie recs failed", "could not parse output as JSON",
"warning", "high", markdown=False)
return 1
recs = data.get("recommendations", [])
if not recs:
push("Movie recs: nothing today",
"The run completed but every candidate was filtered out.",
"warning", "default", markdown=False)
return 0
lines = []
for r in recs:
t = r["tmdb"]
lines.append("**%s** (%s) · %s/10, %sk votes"
% (t["title"], t["year"], t["score"], t["votes"] // 1000))
lines.append("%s" % r.get("reason", "").strip())
lines.append("*from: %s*" % ", ".join(r.get("anchors", [])))
lines.append("")
dropped = (len(data.get("rejected_not_in_tmdb", []))
+ len(data.get("rejected_already_owned", []))
+ len(data.get("rejected_below_min_votes", []))
+ len(data.get("rejected_invented_cluster", [])))
lines.append("_%d shown, %d candidates filtered_" % (len(recs), dropped))
push("%d film recommendations" % len(recs), "\n".join(lines), "clapper")
return 0
if __name__ == "__main__":
sys.exit(main())
hosts/mainframe/bin/rent_utilities
Executable
+527
View File
@@ -0,0 +1,527 @@
#!/usr/bin/python3
"""Assemble the monthly utilities message for the rental and push it to ntfy.
Run monthly from rent-utilities.timer. There are two ways this produces a
number for a bill:
fetch mode -- [imap].enabled, reads the statement mail through Proton
Bridge and regexes the amount out of the body.
template mode -- Bridge not configured yet. Carries the last known figures
and asks you to supply the real ones.
`--set` overrides either, for the months you would rather just read the two
numbers off your phone and be done.
The one rule this script is built around: EVERY run ends in a push. It is
either a message you can forward as-is, or a specific account of what stopped
it. A monthly job that fails quietly is invisible for thirty days, by which
point you have stopped expecting it -- which is the problem this was written
to solve in the first place.
"""
import argparse
import csv
import datetime as dt
import email
import email.policy
import os
import re
import sys
import tomllib
# The Bridge/IMAP/ntfy layer is shared with inbox_tidy. sys.path rather than
# PYTHONPATH so this still works when run by hand, not only from the unit.
sys.path.insert(0, os.path.expanduser("~/.local/lib/pymail"))
from protonimap import ( # noqa: E402
Fatal, imap_connect, imap_date, mailboxes, notify_failure, push,
_body_text, _quote, _sent_at)
CONFIG = os.path.expanduser("~/docs/leases/bills.toml")
CSV_PATH = os.path.expanduser("~/docs/leases/utilities.csv")
OK, MISSING, AMBIGUOUS, ERROR = "ok", "missing", "ambiguous", "error"
# --discover sweeps back from today rather than around a month: it exists to
# find senders, not to resolve a particular month's bill.
DISCOVER_DAYS = 120
# --------------------------------------------------------------------------
# small helpers
# --------------------------------------------------------------------------
def prev_month(today=None):
today = today or dt.date.today()
first = today.replace(day=1)
last_prev = first - dt.timedelta(days=1)
return "%04d-%02d" % (last_prev.year, last_prev.month)
def money(x):
return "%.2f" % x
def month_window(month, slack_days):
"""(since, before) bracketing the month a bill BELONGS to.
Anchored on the requested month, never on today. A plain "last N days"
lookback makes --month a label only: ask for July in September and you get
August's bills filed under July, with nothing looking wrong. The observed
arrival days are AT&T 6th-7th, LG&E 10th-11th, water 14th-15th, so a few
days of slack around the calendar month captures the right bills without
reaching into the neighbouring month's.
"""
y, m = int(month[:4]), int(month[5:7])
first = dt.date(y, m, 1)
nxt = dt.date(y + (m == 12), (m % 12) + 1, 1)
return (first - dt.timedelta(days=slack_days),
nxt + dt.timedelta(days=slack_days))
# --------------------------------------------------------------------------
# ntfy
# --------------------------------------------------------------------------
# --------------------------------------------------------------------------
# CSV
# --------------------------------------------------------------------------
def load_csv():
"""-> {month: {utility: amount}}. Long format, one row per reading."""
out = {}
try:
with open(CSV_PATH, newline="") as fh:
for row in csv.DictReader(fh):
out.setdefault(row["month"], {})[row["utility"]] = float(row["amount"])
except FileNotFoundError:
raise Fatal(
"utilities.csv is missing.\n\n"
"Expected it at %s. If this is a fresh checkout, restore it from "
"utilities.csv.pre-automation in the same directory." % CSV_PATH)
except (KeyError, ValueError) as exc:
raise Fatal(
"utilities.csv could not be parsed: %s\n\n"
"It should be long format with the header `month,utility,amount` "
"and one row per reading. The pre-conversion copy is beside it as "
"utilities.csv.pre-automation." % exc)
return out
def write_csv(table):
rows = sorted(
((m, u, money(a)) for m, us in table.items() for u, a in us.items()),
key=lambda t: (t[0], t[1]))
tmp = CSV_PATH + ".tmp"
with open(tmp, "w", newline="") as fh:
# lineterminator is explicit: csv.writer defaults to CRLF, which would
# quietly convert a file that has always been LF.
w = csv.writer(fh, lineterminator="\n")
w.writerow(["month", "utility", "amount"])
w.writerows(rows)
os.replace(tmp, CSV_PATH)
# --------------------------------------------------------------------------
# IMAP / Proton Bridge
# --------------------------------------------------------------------------
def message_bodies(conn, imap_cfg, sender, since, before):
"""Most-recent-first list of (subject, body-text) from `sender`.
Ordered by the Date header across every mailbox searched, so "most
recent" still means most recent when a bill was filed out of INBOX.
"""
found = []
for box in mailboxes(imap_cfg):
typ, _ = conn.select(_quote(box), readonly=True)
if typ != "OK":
continue # folder renamed or gone; the others still count
typ, data = conn.search(
None, '(SINCE "%s" BEFORE "%s" FROM "%s")'
% (imap_date(since), imap_date(before), sender))
if typ != "OK":
continue
for num in data[0].split():
typ, raw = conn.fetch(num, "(RFC822)")
if typ != "OK" or not raw or not raw[0]:
continue
msg = email.message_from_bytes(raw[0][1], policy=email.policy.default)
found.append((_sent_at(msg),
str(msg.get("Subject", "(no subject)")),
_body_text(msg)))
found.sort(key=lambda t: t[0], reverse=True)
return [(subj, body) for _, subj, body in found]
def mailbox_total(conn, imap_cfg, since, before):
"""How many messages exist in the window at all, from anyone.
Distinguishes "the bill did not arrive" from "Bridge has not finished
syncing yet". A first sync takes ~90 minutes, and a partially-synced
mailbox answers searches successfully with incomplete results -- so a
missing bill and an unsynced mailbox look identical unless this is
checked. Reporting the wrong one would send you hunting for a sender that
was right all along.
"""
total = 0
try:
for box in mailboxes(imap_cfg):
typ, _ = conn.select(_quote(box), readonly=True)
if typ != "OK":
continue
typ, data = conn.search(None, '(SINCE "%s" BEFORE "%s")'
% (imap_date(since), imap_date(before)))
if typ == "OK":
total += len(data[0].split())
return total
except Exception:
return None
def extract_amount(bodies, pattern):
"""-> (status, value, detail)."""
rx = re.compile(pattern)
# Every matching message in the window is considered, not just the newest.
# Taking the newest would quietly resolve a window that had slipped and
# caught two months of bills -- the failure this must not have. Within a
# correctly scoped month there is exactly one bill, so two DIFFERENT
# figures means something is wrong and is worth refusing. The same figure
# repeated is fine: LG&E sends the identical mail twice, and AT&T's
# "Bill total" appears once per bill.
found, where = {}, {}
for subject, body in bodies:
for m in rx.finditer(body):
v = m.group(1).replace(",", "")
found[v] = found.get(v, 0) + 1
where.setdefault(v, subject)
if len(found) == 1:
v = next(iter(found))
return OK, float(v), where[v]
if len(found) > 1:
vals = ", ".join("$%s (%r)" % (v, where[v][:44]) for v in sorted(found))
return AMBIGUOUS, None, (
"matched %d DIFFERENT amounts in the window -- %s. Refusing to "
"guess which is this month's." % (len(found), vals))
if bodies:
return MISSING, None, (
"%d message(s) from that sender, but the pattern matched no dollar "
"amount in any of them. Most recent was %r"
% (len(bodies), bodies[0][0]))
return MISSING, None, "no message from that sender in the search window"
# --------------------------------------------------------------------------
# rendering
# --------------------------------------------------------------------------
def render_message(cfg, amounts):
"""The text to forward to them. Layout matches the original utilities.py.
One deliberate difference: the original's Total line printed `$$` because
total_amount already carried a '$' and the f-string added another.
"""
split = cfg["split"]
lines = []
full = their = 0.0
for bill in cfg["bills"]:
name = bill["name"]
amt = amounts[name]
share = amt * split
full += amt
their += share
lines.append("%-15s $%-8s Your Share = $%s"
% (name + ":", money(amt) + ",", money(share)))
lines.append("%-15s $%-8s Your Total Share = $%s"
% ("Total:", money(full) + ",", money(their)))
body = "\n".join(lines)
greeting = cfg.get("greeting", "").strip()
return (greeting + "\n\n" + body) if greeting else body
def render_problems(month, problems, note=None):
out = ["**Could not build the %s message.**" % month, ""]
if note:
out += [note, ""]
for name, detail in problems:
out.append("**%s**" % name)
out.append(detail)
out.append("")
out.append("---")
out.append("Nothing was written to utilities.csv. Once you have the real "
"figures, run:")
out.append("")
out.append("```")
# Join the continuations explicitly rather than appending a trailing "\\"
# to every line -- the last one must NOT have it, or pasting the command
# leaves the shell waiting on a continuation that never comes.
cmd = ["rent_utilities --month %s" % month]
cmd += [" --set '%s=0.00'" % name for name, _ in problems]
out.append(" \\\n".join(cmd))
out.append("```")
return "\n".join(out)
# --------------------------------------------------------------------------
# modes
# --------------------------------------------------------------------------
def discover(cfg):
imap_cfg = cfg["imap"]
conn = imap_connect(imap_cfg)
try:
conn.select(_quote(mailboxes(imap_cfg)[0]), readonly=True)
since = dt.date.today() - dt.timedelta(days=DISCOVER_DAYS)
typ, data = conn.search(None, '(SINCE "%s")' % imap_date(since))
if typ != "OK":
raise Fatal("IMAP search failed: %s" % typ)
ids = data[0].split()
print("%d messages since %s\n" % (len(ids), since))
rx = re.compile(r"\$\s*([\d,]+\.\d{2})")
for num in reversed(ids):
typ, raw = conn.fetch(num, "(RFC822)")
if typ != "OK" or not raw or not raw[0]:
continue
msg = email.message_from_bytes(raw[0][1], policy=email.policy.default)
body = ""
try:
part = msg.get_body(preferencelist=("plain", "html"))
body = part.get_content() if part is not None else ""
except Exception:
pass
body = re.sub(r"<[^>]+>", " ", body)
amounts = sorted({m.group(1) for m in rx.finditer(body)})
print("from: %s" % msg.get("From", "?"))
print("subject: %s" % msg.get("Subject", "?"))
print("date: %s" % msg.get("Date", "?"))
print("amounts: %s" % (", ".join("$" + a for a in amounts[:8]) or "none"))
print()
finally:
try:
conn.logout()
except Exception:
pass
return 0
def collect(cfg, month, overrides):
"""-> (amounts, problems, used_fetch)."""
amounts, problems = {}, []
imap_cfg = cfg.get("imap", {})
use_imap = imap_cfg.get("enabled", False)
conn = None
since, before = month_window(month, imap_cfg.get("window_slack_days", 5))
try:
for bill in cfg["bills"]:
name = bill["name"]
if name in overrides:
amounts[name] = overrides[name]
continue
if "fixed" in bill:
amounts[name] = float(bill["fixed"])
continue
if not use_imap:
problems.append((name, "no amount supplied, and [imap].enabled "
"is false so nothing was fetched."))
continue
sender, pattern = bill.get("sender", ""), bill.get("pattern", "")
if not sender or not pattern:
problems.append((name, (
"not configured for fetching: `sender` and/or `pattern` "
"are empty in bills.toml. Run `rent_utilities --discover` "
"to see the senders and candidate amounts in your recent "
"mail, then fill them in.")))
continue
if conn is None:
conn = imap_connect(imap_cfg)
try:
bodies = message_bodies(conn, imap_cfg, sender, since, before)
except Fatal:
raise
except Exception as exc:
problems.append((name, "IMAP read failed: %r" % (exc,)))
continue
status, value, detail = extract_amount(bodies, pattern)
if status == OK:
amounts[name] = value
continue
note = ""
if not bodies:
total = mailbox_total(conn, imap_cfg, since, before)
if total == 0:
note = ("\n\n**The mailbox returned NO messages at all in "
"this window, from anyone.** That is almost "
"certainly Bridge still syncing rather than a "
"missing bill -- a first sync takes ~90 minutes. "
"Check with:\n"
" sudo podman logs --tail 20 "
"connor_protonmail-bridge_1\n"
"Do not go changing the sender in bills.toml yet.")
elif total is not None:
note = ("\n\n(%d message(s) from other senders were found "
"in the same window, so the mailbox is reachable "
"and this sender genuinely has nothing.)" % total)
problems.append((name, "%s\n\nSearched for `%s` between %s and %s in %s.%s"
% (detail, sender, since, before,
", ".join(mailboxes(imap_cfg)), note)))
finally:
if conn is not None:
try:
conn.logout()
except Exception:
pass
return amounts, problems, use_imap
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--month", help="target month as YYYY-MM (default: last month)")
ap.add_argument("--set", action="append", default=[], metavar="NAME=AMOUNT",
help="supply an amount by hand; repeatable")
ap.add_argument("--discover", action="store_true",
help="print recent senders and candidate amounts, then exit")
ap.add_argument("--dry-run", action="store_true",
help="print what would be pushed instead of pushing")
args = ap.parse_args()
# Config load is outside the push-on-failure net: without it there is no
# ntfy url or topic to push to. It surfaces on stderr, and systemd will
# mark the unit failed.
try:
with open(CONFIG, "rb") as fh:
cfg = tomllib.load(fh)
except (OSError, tomllib.TOMLDecodeError) as exc:
print("rent_utilities: cannot read config %s: %s" % (CONFIG, exc),
file=sys.stderr)
return 2
if args.discover:
try:
return discover(cfg)
except Fatal as exc:
print("rent_utilities: %s" % exc, file=sys.stderr)
return 1
month = args.month or prev_month()
if not re.fullmatch(r"\d{4}-\d{2}", month):
print("rent_utilities: --month must be YYYY-MM, got %r" % month,
file=sys.stderr)
return 2
overrides = {}
for item in args.set:
if "=" not in item:
print("rent_utilities: --set wants NAME=AMOUNT, got %r" % item,
file=sys.stderr)
return 2
k, v = item.split("=", 1)
try:
overrides[k.strip()] = float(v)
except ValueError:
print("rent_utilities: %r is not a number in %r" % (v, item),
file=sys.stderr)
return 2
known = {b["name"] for b in cfg["bills"]}
unknown = set(overrides) - known
if unknown:
print("rent_utilities: --set names not in bills.toml: %s (known: %s)"
% (", ".join(sorted(unknown)), ", ".join(sorted(known))),
file=sys.stderr)
return 2
try:
table = load_csv()
amounts, problems, used_fetch = collect(cfg, month, overrides)
except Fatal as exc:
body = "**%s could not run.**\n\n%s" % (month, exc)
if args.dry_run:
print(body)
else:
notify_failure(cfg, "Rent utilities: FAILED", body)
return 1
if problems:
note = None
if not used_fetch:
last = max((m for m in table if m < month), default=None)
if last:
prev = ", ".join(
"%s $%s" % (u, money(a)) for u, a in sorted(table[last].items()))
note = ("Template mode -- Proton Bridge is not wired up yet, so "
"nothing was fetched.\n\nLast figures on file (%s): %s"
% (last, prev))
body = render_problems(month, problems, note)
title = ("Rent utilities: %s needs numbers" % month if not used_fetch
else "Rent utilities: %s incomplete" % month)
if args.dry_run:
print(body)
else:
notify_failure(cfg, title, body)
return 1
msg = render_message(cfg, amounts)
total = sum(amounts.values()) * cfg["split"]
body = ("Their share for %s: **$%s**\n\nForward this:\n\n```\n%s\n```"
% (month, money(total), msg))
# Before the write, not after: a dry run must not touch utilities.csv.
if args.dry_run:
print(body)
return 0
table.setdefault(month, {}).update(amounts)
try:
write_csv(table)
except OSError as exc:
body = ("**Built the %s figures but could not save them.**\n\n"
"Writing %s failed: %s\n\nThe message below is still correct, "
"but it was NOT recorded -- next month will not have this "
"month's history.\n\n```\n%s\n```"
% (month, CSV_PATH, exc, render_message(cfg, amounts)))
# No dry-run branch here: a dry run has already returned above, so
# reaching this point means the write was real and really failed.
notify_failure(cfg, "Rent utilities: %s not saved" % month, body)
return 1
try:
push(cfg, "Rent utilities: %s" % month, body, "house,moneybag")
except (urllib.error.URLError, OSError) as exc:
print("rent_utilities: built the message but ntfy push failed: %s"
% exc, file=sys.stderr)
print(msg)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())