""" Sopel plugin: twtxt feed Commands (owner only): !twt - show usage !twt last [n] - show last n twts, most recent first (default 5) !twt count - total number of twts !twt delete - delete the most recent twt (asks yes/no) !twt yes / !twt no - confirm or cancel a pending delete !twt - post a twt Config ([twt] section in sopel config): twtxt_path - path to twtxt.txt (default /var/www/twt.gumx.cc/twtxt.txt) timezone - timezone name for timestamps (default UTC) nick - nick value written in file header (default gumx) url - url value written in file header """ import glob import os from datetime import datetime, timezone from pathlib import Path from zoneinfo import ZoneInfo from sopel import plugin from sopel.config.types import FilenameAttribute, StaticSection, ValidatedAttribute import os as _os, sys as _sys _alfred_dir = _os.path.dirname(_os.path.abspath(__file__)) if _alfred_dir not in _sys.path: _sys.path.insert(0, _alfred_dir) import helpstrings as h _pending_delete = {} _SUBCOMMANDS = {"last", "count", "delete", "yes", "no", "add", "users", "rm"} WKFO_TOKENS = '/etc/nginx/twt-tokens.map' WKFO_TWR = '/var/www/wk.fo/twt' def _load_twt_tokens(): tokens = {} try: with open(WKFO_TOKENS) as f: for line in f: line = line.strip() if not line or line.startswith('#') or not line.startswith('"'): continue end = line.index('"', 1) tok = line[1:end] rest = line[end+1:].strip().rstrip(';').split('#')[0].strip() if rest: tokens[tok] = rest except FileNotFoundError: pass return tokens def _save_twt_tokens(tokens): with open(WKFO_TOKENS, 'w') as f: for tok, user in tokens.items(): f.write(f'"{tok}" {user};\n') def _esc(s): return s.replace('&', '&').replace('<', '<').replace('>', '>').replace('"', '"') def _load_twtxt(path): twts = [] try: with open(path) as f: for line in f: line = line.strip() if not line or line.startswith('#'): continue parts = line.split('\t', 1) if len(parts) == 2: try: dt = datetime.fromisoformat(parts[0].replace('Z', '+00:00')) except ValueError: dt = datetime.min.replace(tzinfo=timezone.utc) twts.append((dt, parts[1])) except FileNotFoundError: pass return twts _WKFO_HEAD = """\ {title} """ _WKFO_FOOTER = """\ """ def _build_user_page(username): twts = _load_twtxt(f'{WKFO_TWR}/~{username}/twtxt.txt') items = '\n'.join( f'
  • [{dt.strftime("%Y-%m-%d")}] {_esc(text)}
    • ' for dt, text in reversed(sorted(twts)) ) or '
  • (no twts yet)
    • ' return ( _WKFO_HEAD.format(title=f'wk.fo / twt / ~{_esc(username)}') + '\n' '
    \n' f'\n' '
    \n' '
    \n' f'

    ~{_esc(username)}

    \n' f'

    twtxt feed

    \n' '
    \n' '

    \n' '

    \n' '

    \n' '

    \n' '
    \n' '\n' '
    \n' '\n' '
    \n' + _WKFO_FOOTER ) def _build_index_page(): users = sorted( os.path.basename(d)[1:] for d in glob.glob(f'{WKFO_TWR}/~*') if os.path.isdir(d) ) all_twts = [] for u in users: for dt, text in _load_twtxt(f'{WKFO_TWR}/~{u}/twtxt.txt'): all_twts.append((dt, u, text)) all_twts.sort(reverse=True) feed_items = '\n'.join( f'
  • [{dt.strftime("%Y-%m-%d")}] ' f'~{_esc(u)}: {_esc(text)}
    • ' for dt, u, text in all_twts[:50] ) or '
  • (nothing yet)
    • ' user_items = '\n'.join( f'
  • ~{_esc(u)} ' f'(twtxt)
    • ' for u in users ) or '
  • (no users yet)
    • ' return ( _WKFO_HEAD.format(title='wk.fo / twt') + '\n' '
    \n' '\n' '
    \n' '
    \n' '

    twt

    \n' '

    Hosted twtxt. Short messages from friends.

    \n' '

    feed

    \n' '\n' '

    users

    \n' '\n' '
    \n' + _WKFO_FOOTER ) def _wkfo_regen(bot, username=None): try: root = Path(WKFO_TWR) root.mkdir(parents=True, exist_ok=True) if username: Path(f'{WKFO_TWR}/~{username}/index.html').write_text( _build_user_page(username), encoding='utf-8') Path(f'{WKFO_TWR}/index.html').write_text( _build_index_page(), encoding='utf-8') except Exception as e: bot.say(f'regen error: {e}') class TwtSection(StaticSection): twtxt_path = FilenameAttribute("twtxt_path", relative=False, default="/var/www/twt.gumx.cc/twtxt.txt") timezone = ValidatedAttribute("timezone", default="UTC") nick = ValidatedAttribute("nick", default="gumx") url = ValidatedAttribute("url", default="https://twt.gumx.cc/twtxt.txt") def setup(bot): bot.settings.define_section("twt", TwtSection) def _is_owner(bot, trigger): return trigger.nick == bot.settings.core.owner def _path(bot): return bot.settings.twt.twtxt_path def _read(bot): p = _path(bot) if not os.path.exists(p): return [] with open(p) as f: lines = f.readlines() twts = [] for line in lines: line = line.rstrip("\n") if not line or line.startswith("#"): continue tab = line.find("\t") if tab == -1: continue twts.append((line[:tab], line[tab + 1:])) return twts def _write(bot, twts): p = _path(bot) nick = bot.settings.twt.nick url = bot.settings.twt.url with open(p, "w") as f: f.write(f"# nick = {nick}\n") f.write(f"# url = {url}\n") for ts, text in twts: f.write(f"{ts}\t{text}\n") def _append(bot, text): tz = ZoneInfo(bot.settings.twt.timezone) now = datetime.now(tz) offset = now.strftime("%z") offset = offset[:3] + ":" + offset[3:] ts = now.strftime("%Y-%m-%dT%H:%M:%S") + offset p = _path(bot) nick = bot.settings.twt.nick url = bot.settings.twt.url if not os.path.exists(p): os.makedirs(os.path.dirname(p), exist_ok=True) with open(p, "w") as f: f.write(f"# nick = {nick}\n") f.write(f"# url = {url}\n") with open(p, "a") as f: f.write(f"{ts}\t{text}\n") return ts @plugin.commands("twt") def cmd_twt(bot, trigger): if not _is_owner(bot, trigger): return raw = (trigger.group(2) or "").strip() args = raw.split() sub = args[0].lower() if args else "" if not sub: bot.say(h.TWF_TOPIC) return # --- last --- if sub == "last": try: n = int(args[1]) if len(args) > 1 else 5 n = max(1, min(n, 10)) except ValueError: bot.say("usage: !twt last [n]") return twts = _read(bot) if not twts: bot.say("no twts yet.") return for ts, text in reversed(twts[-n:]): bot.say(f"[{ts[:10]}] {text}") return # --- count --- if sub == "count": n = len(_read(bot)) bot.say(f"{n} twt{'s' if n != 1 else ''}") return # --- delete --- if sub == "delete": twts = _read(bot) if not twts: bot.say("nothing to delete.") return ts, text = twts[-1] _pending_delete[trigger.nick] = True bot.say(f"delete [{ts[:10]}] {text!r}? !twt yes / !twt no") return # --- yes/no (confirm delete) --- if sub == "yes": if not _pending_delete.pop(trigger.nick, False): bot.say("nothing pending.") return twts = _read(bot) if not twts: bot.say("nothing to delete.") return removed = twts.pop() _write(bot, twts) bot.say(f"deleted [{removed[0][:10]}] {removed[1]!r}") return if sub == "no": if _pending_delete.pop(trigger.nick, False): bot.say("cancelled.") return # --- wk.fo/twt user management --- if sub == "users": tokens = _load_twt_tokens() if not tokens: bot.say("no wk.fo/twt users") return users = sorted(set(tokens.values())) bot.say(", ".join(users)) return if sub == "add": if len(args) < 3: bot.say("usage: !twt add ") return username = args[1] token = args[2] tokens = _load_twt_tokens() if token in tokens: bot.say(f"token already exists for {tokens[token]}") return user_dir = Path(f'{WKFO_TWR}/~{username}') user_dir.mkdir(parents=True, exist_ok=True) twtxt = user_dir / 'twtxt.txt' if not twtxt.exists(): twtxt.write_text(f'# nick = {username}\n# url = https://wk.fo/twt/~{username}/twtxt.txt\n') tokens[token] = username _save_twt_tokens(tokens) _wkfo_regen(bot, username) bot.say(f"added {username}") return if sub == "rm": if len(args) < 2: bot.say("usage: !twt rm ") return target = args[1] tokens = _load_twt_tokens() before = len(tokens) tokens = {t: u for t, u in tokens.items() if u != target} if len(tokens) == before: bot.say(f"not found: {target}") return _save_twt_tokens(tokens) _wkfo_regen(bot) bot.say(f"removed {target} (feed files kept)") return # --- post (anything that's not a subcommand) --- ts = _append(bot, raw) bot.say(f"twted [{ts[:10]}]: {raw}")