import on_this_day
import sys
import sqlite3
from datetime import datetime
from config import config
from http.server import HTTPServer, BaseHTTPRequestHandler
def format_datetime(date):
return str(date)
def generate_item(item):
return f"""-
{config["host"]}/{item["guid"]}
{item["guid"]}
{item["createdAt"]}
"""
def generate_feed(items):
return f"""
On This Day
Daily posts of stuff for today
{format_datetime(datetime.now())}
{"".join(generate_item(item) for item in items)}
"""
def generate_ul(items):
return f"""
{f"- {'
- '.join(items)}
"}
"""
def get_description():
list_items = [
(on_this_day.get_old_news, "old news"),
(on_this_day.get_peanuts, "peanuts"),
(on_this_day.get_calvin_and_hobbes, "calvin and hobbes"),
(on_this_day.get_today_wikipedia, "wikipedia"),
(on_this_day.get_week_holidays, "holiday"),
(on_this_day.get_crossword, "crossword"),
(on_this_day.year_progress_bar, "year progress"),
(on_this_day.get_homepage_links, "homepage links"),
]
items = []
for func, name in list_items:
try:
items.append(func())
except Exception as e:
items.append(f"Error getting {name}")
print("Exception:", e)
ul = generate_ul(items)
div_items = [
(on_this_day.get_today_wikiquote, "wikiquote"),
(on_this_day.get_thoreau, "thoreau"),
]
items = []
for func, name in div_items:
try:
items.append(func())
except:
items.append(f"Error getting {name}")
divs = f"{'
'.join(items)}
"
return f"""
On this day
{ul}
{divs}
"""
def setup(con):
cur = con.cursor()
cur.execute("CREATE TABLE IF NOT EXISTS items (guid, title, description, createdAt)")
con.commit()
cur.close()
def insert_today():
con = sqlite3.connect(config["db"])
now = datetime.now()
cur = con.cursor()
cur.execute(
"INSERT INTO items values (?, ?, ?, ?)",
(now.isoformat(), str(now), get_description(), format_datetime(now))
)
con.commit()
cur.close()
def get_all(con):
cur = con.cursor()
items = []
for (guid, title, description, createdAt) in cur.execute("SELECT guid, title, description, createdAt FROM items"):
items.append({
"guid": guid,
"title": title,
"description": description,
"createdAt": createdAt,
})
cur.close()
return generate_feed(items)
def get_one_desc(con, guid):
cur = con.cursor()
description = next(
cur.execute("SELECT description FROM items WHERE guid=:guid", {"guid": guid}))[0]
cur.close()
print(description)
return description
class RssHTTPRequestHandler(BaseHTTPRequestHandler):
def __init__(self, *args):
BaseHTTPRequestHandler.__init__(self, *args)
def do_GET(self):
if self.path == "/":
self.send_response(200)
self.end_headers()
self.wfile.write(bytes(get_all(con), "utf-8"))
else:
guid = self.path[1:]
self.send_response(200)
self.end_headers()
self.wfile.write(bytes(get_one_desc(con, guid), "utf-8"))
con = None
def server():
print("Starting http server")
http = HTTPServer(("", 8000), RssHTTPRequestHandler)
print("serving forever")
http.serve_forever()
def test():
print(get_description())
if __name__ == "__main__":
if len(sys.argv) == 1:
con = sqlite3.connect(config["db"])
setup(con)
server()
elif sys.argv[1] == "insert":
insert_today()
elif sys.argv[1] == "test":
test()