#!/usr/bin/env python3

"""

q2n - QEC to NNTP sync

This script syncs QEC logs to NNTP.

• everything configurable via `Config`

• has a throttler so we don't accidentally submit too much at a time

• has a dry-run for submission

• it remembers what has already been submitted so they don't get submitted

again

• TODO:

- read from argv or a config file

- set up a cron job

- put it on tildegit (once my application issue get sorted out)

"""

from dataclasses import dataclass

import datetime as dt

import io

import logging

import os

import pickle

import pwd

import random

import subprocess as sp

import time

import typing as t

_LOGGER = logging.getLogger(__name__)

Path = str

User = str

NntpArticleBody = str

LogEntryHash = str

@dataclass

class Config:

listing_dir: str

listing_filename: str

nntp_group: str

nntp_server: str

max_submission: int

submission_store_dir: Path

@classmethod

def create(cls):

return Config(

listing_dir="/var/gopher/",

listing_filename="listing.gophermap",

nntp_server="localhost",

TODO: find more appropriate one

nntp_group="cosmic.worldbuilding",

max_submission=5,

submission_store_dir="/var/tmp/q2n",

)

@dataclass

class Ship:

name: str

owner: User

@dataclass

class LogEntry:

ship: Ship

author: User

title: str

file_name: str

class LogIterator(t.Protocol):

def __call__(self) -> t.List[LogEntry]: ...

class SubmitCondition(t.Protocol):

def __call__(self, log_entry: LogEntry) -> bool: ...

class LogSubmitter(t.Protocol):

def __call__(self, log: LogEntry) -> None: ...

@dataclass

class Utils:

config: Config

def ship_owner(self, ship_name: str) -> User:

return self._get_path_user(

f"{self.config.listing_dir}/{ship_name}"

)

def read_log_content(self, log: LogEntry) -> str:

return self._read_log_entry(

f"{self.config.listing_dir}/{log.ship.name}/{log.file_name}"

)

@staticmethod

def _read_log_entry(path: str) -> str:

with open(path, "r", encoding="utf-8") as f:

return f.read()

@staticmethod

def _get_path_user(fp: str) -> User:

st = os.stat(fp)

return pwd.getpwuid(st.st_uid).pw_name

@dataclass

class SubmittedLogsStore:

store_dir: str

def __post_init__(self):

import subprocess as sp

sp.check_call(

f"mkdir -p {self.store_dir}",

shell=True

)

def record_submission(self, log: LogEntry):

with open(f"{self.store_dir}/{self.checksum(log)}", "wb") as f:

pickle.dump(log, f)

def load_submitted_logs(self) -> t.List[LogEntryHash]:

return os.listdir(self.store_dir)

@staticmethod

def checksum(log: LogEntry) -> LogEntryHash:

import hashlib

checked_str = f"{log.ship.name}{log.file_name}"

return hashlib.md5(checked_str.encode("utf-8")).hexdigest()

Throttles log entries to submit. Just in case there's a bug.

Usually we'd limit logs to submit to a small number, and maybe also

send out some alert.

SubmissionThrottle = t.Callable[[t.List[LogEntry]], t.List[LogEntry]]

@dataclass

class ListingFileLogIterator(LogIterator):

listing_dir: str

listing_filename: str

utils: Utils

def __call__(self) -> t.List[LogEntry]:

with open(

f"{self.listing_dir}/{self.listing_filename}",

"r",

encoding="utf-8"

) as f:

entries = f.readlines()

return [self._parse(ent) for ent in entries]

def _parse(self, entry: str) -> LogEntry:

"""Parse a listing file entry into a `LogEntry`

An entry looks like this:

0betsy - About QEC /betsy/qec.txt

I.e.

0 -