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

Refactor OVER, HDR

Browse source 
Refactor OVER, HDR to be simpler, using a single convenience method to
implement similar semantics for commands which operate on both single
and ranges of messages
This commit is contained in:
XANTRONIX Development 2024-11-26 14:53:16 -05:00
parent 72a4fd2230
commit 65a60d33fd
1 changed files with 60 additions and 83 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

137
lib/nntp/tiny/session.py
View file

@ -518,17 +518,62 @@ class Session():
return self.end()
def _message_by_id(self, identifier: str):
if identifier[0] == '<':
return self.db.get(Message, {
'newsgroup_id': self.newsgroup.id,
def _each_message_by_id(self, identifier: str):
if identifier is None:
if self.newsgroup is None:
self.respond(ResponseCode.NNTP_NEWSGROUP_NOT_SELECTED)
return
if self.article_id is None:
self.respond(ResponseCode.NNTP_ARTICLE_INVALID_NUMBER)
return
message = self.db.get(Message, self.article_id)
if message is None:
self.respond(ResponseCode.NNTP_ARTICLE_INVALID_NUMBER)
return
self.respond(ResponseCode.NNTP_INFORMATION_FOLLOWS)
yield message
elif identifier[0] == '<':
message = self.db.query(Message, {
'message_id': identifier
})
}).fetchone()
if message is None:
self.respond(ResponseCode.NNTP_ARTICLE_NOT_FOUND_ID)
return
self.respond(ResponseCode.NNTP_INFORMATION_FOLLOWS)
yield message
else:
return self.db.get(Message, {
'newsgroup_id': self.newsgroup.id,
'id': int(identifier)
})
if self.newsgroup is None:
self.respond(ResponseCode.NNTP_NEWSGROUP_NOT_SELECTED)
return
msgrange = MessageRange.parse(identifier)
sql = f"select * from {Message.name} where "
sql += " newsgroup_id = ? and " + msgrange.where()
cr = self.db.query_sql(Message, sql, (self.newsgroup.id,))
first = True
for message in cr.each():
if first:
first = False
self.respond(ResponseCode.NNTP_INFORMATION_FOLLOWS)
yield message
if first:
self.respond(ResponseCode.NNTP_ARTICLE_NOT_FOUND_RANGE)
self.end()
def _send_message_headers(self, message: Message):
for name in message.headers:
@ -587,39 +632,9 @@ class Session():
message.id, message.headers.get(name, '')
))
def _cmd_hdr(self, name: str, msg: Optional[str]=None):
if self.newsgroup is None:
return self.respond(ResponseCode.NNTP_NEWSGROUP_NOT_SELECTED)
if msg is None:
if self.article_id is None:
return self.respond(ResponseCode.NNTP_ARTICLE_INVALID_NUMBER)
message = self.db.get(Message, self.article_id)
self.respond(ResponseCode.NNTP_HEADERS_FOLLOW)
def _cmd_hdr(self, name: str, identifier: Optional[str]=None):
for message in self._each_message_by_id(identifier):
self._send_message_header(message, name)
else:
msgrange = MessageRange.parse(msg)
sql = f"select * from {Message.name} where "
sql += " newsgroup_id = ? and " + msgrange.where()
cr = self.db.query_sql(Message, sql, (self.newsgroup.id,))
first = True
for message in cr.each():
if first:
first = False
self.respond(ResponseCode.NNTP_HEADERS_FOLLOW)
self._send_message_header(message, name)
if first:
return self.respond(ResponseCode.NNTP_ARTICLE_NOT_FOUND_RANGE)
return self.end()
def _message_overview(self, message: Message) -> dict:
return map(lambda s: s.replace('\t', ' '), [
@ -634,49 +649,11 @@ class Session():
])
def _cmd_over(self, identifier: Optional[str]=None):
if identifier is None:
if self.newsgroup is None:
return self.respond(ResponseCode.NNTP_NEWSGROUP_NOT_SELECTED)
if self.article_id is None:
return self.respond(ResponseCode.NNTP_ARTICLE_INVALID_NUMBER)
message = self.db.get(Message, self.article_id)
if message is None:
return self.respond(ResponseCode.NNTP_ARTICLE_INVALID_NUMBER)
elif identifier[0] == '<':
message = self.db.query(Message, {
'message_id': identifier
}).fetchone()
if message is None:
return self.respond(ResponseCode.NNTP_ARTICLE_NOT_FOUND_ID)
else:
if self.newsgroup is None:
return self.respond(ResponseCode.NNTP_NEWSGROUP_NOT_SELECTED)
msgrange = MessageRange.parse(identifier)
sql = f"select * from {Message.name} where "
sql += " newsgroup_id = ? and " + msgrange.where()
cr = self.db.query_sql(Message, sql, (self.newsgroup.id,))
first = True
for message in cr.each():
if first:
first = False
self.respond(ResponseCode.NNTP_OVERVIEW_FOLLOWS)
for message in self._each_message_by_id(identifier):
overview = self._message_overview(message)
self.print('|'.join(overview))
if first:
return self.respond(ResponseCode.NNTP_ARTICLE_NOT_FOUND_RANGE)
def _cmd_stat(self, identifier: Optional[str]=None):
if self.newsgroup is None:
return self.respond(ResponseCode.NNTP_NEWSGROUP_NOT_SELECTED)