xan/xenu_nntp
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Initial implementation of lib/nntp/tiny/mbox.py

Browse source
This commit is contained in:
XANTRONIX Development 2024-11-07 11:39:19 -05:00
parent fdac039c2a
commit 240a79e3ba
1 changed files with 26 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
ingest.py → lib/nntp/tiny/mbox.py
View file

@ -1,7 +1,5 @@
#! /usr/bin/env python3
import sys
import re import re
import enum
import sqlite3 import sqlite3
class MBoxReaderError(Exception): class MBoxReaderError(Exception):
@ -58,18 +56,29 @@ class MBoxReaderBuffer():
and self.is_header_line(3): and self.is_header_line(3):
return 3 return 3
class MBoxMessageState(enum.Enum):
EMPTY = 0
HEADER = 1
BODY = 2
class MBoxMessage(): class MBoxMessage():
__slots__ = 'headers', 'body', 'key', __slots__ = 'state', 'headers', 'line', 'body', 'key',
def __init__(self): def __init__(self):
self.state = MBoxMessageState.EMPTY
self.headers = dict() self.headers = dict()
self.body = None self.line = None
self.body = ''
self.key = None self.key = None
def add(self, line: str): def add(self, line: str):
if self.body is None: if self.state is MBoxMessageState.EMPTY:
if line == '\n': self.state = MBoxMessageState.HEADER
self.body = ''
if self.state is MBoxMessageState.HEADER:
if line == '\n' or line == '\r\n':
self.state = MBoxMessageState.BODY
self.body = ''
elif line[0] == ' ' or line[0] == '\t': elif line[0] == ' ' or line[0] == '\t':
self.headers[self.key] += ' ' + line.strip() self.headers[self.key] += ' ' + line.strip()
else: else:
@ -78,15 +87,17 @@ class MBoxMessage():
if match: if match:
self.key = match[1] self.key = match[1]
self.headers[self.key] = line.rstrip() self.headers[self.key] = match[2].rstrip()
else: elif self.state is MBoxMessageState.BODY:
self.body += line if self.line is None:
self.body = line
else:
self.body += self.line
def is_empty(self): self.line = line
return len(self.headers) == 0 and self.body is None
def is_first_line(self): def is_first_line(self):
return len(self.headers) == 1 and self.body is None return len(self.headers) == 1 and self.body == ''
class MBoxReader(): class MBoxReader():
__slots__ = 'path', 'fh', 'line', 'buf', 'message', __slots__ = 'path', 'fh', 'line', 'buf', 'message',
@ -102,7 +113,7 @@ class MBoxReader():
while True: while True:
line = self.fh.readline() line = self.fh.readline()
if line is None: if line is None or line == '':
ret = self.message ret = self.message
self.message = None self.message = None
@ -135,13 +146,3 @@ class MBoxReader():
break break
yield message yield message
db = sqlite3.connect(sys.argv[1])
reader = MBoxReader(sys.argv[2])
count = 0
for message in reader.messages():
count += 1
print(f"Found {count} messages")