261 lines
8.9 KiB
Python
261 lines
8.9 KiB
Python
import json
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import ox
|
|
|
|
from django.utils import timezone
|
|
from django.utils.timezone import datetime, timedelta
|
|
from django.shortcuts import render, redirect
|
|
from django.db.models import Q
|
|
from django.utils.html import mark_safe
|
|
from django.conf import settings
|
|
from django.http import HttpResponse, Http404
|
|
|
|
from . import models
|
|
from . import tasks
|
|
from ..signalbot.rpc import send_reaction
|
|
from .utils import render_to_json
|
|
from ..utils import default_context
|
|
|
|
TS_FORMAT = "%Y-%m-%dT%H:%M:%S"
|
|
|
|
|
|
def get_now():
|
|
return timezone.make_aware(datetime.now(), timezone.get_default_timezone())
|
|
|
|
|
|
def get_monday(date):
|
|
d = date.strftime('%Y-W%W-1')
|
|
return datetime.strptime(d, "%Y-W%W-%w").strftime('%Y-%m-%d')
|
|
|
|
|
|
def format_week(week):
|
|
a = datetime.strptime(week, '%Y-%m-%d')
|
|
b = (a + timedelta(days=6))
|
|
fmt = '%b %d'
|
|
a = a.strftime(fmt)
|
|
b = b.strftime(fmt)
|
|
extra = models.Week.objects.filter(monday=week).first()
|
|
if extra and extra.title:
|
|
extra = ': ' + extra.title
|
|
else:
|
|
extra = ''
|
|
return '%s - %s%s' % (a, b, extra)
|
|
|
|
def get_byline(week):
|
|
extra = models.Week.objects.filter(monday=week).first()
|
|
if extra:
|
|
return extra.byline
|
|
|
|
def get_weeks(archive):
|
|
weeks = sorted(set(archive.values_list('year', 'week')))
|
|
weeks = [datetime.strptime('%s-W%s-1' % w, "%Y-W%W-%w").strftime('%Y-%m-%d') for w in weeks]
|
|
weeks = [{
|
|
'date': week,
|
|
'year': week.split('-')[0],
|
|
'title': format_week(week)
|
|
} for week in weeks]
|
|
return weeks
|
|
|
|
|
|
def index(request):
|
|
context = default_context(request)
|
|
now = request.GET.get("now")
|
|
if request.user.is_staff and now:
|
|
now = datetime.strptime(now, TS_FORMAT)
|
|
now = timezone.make_aware(now, timezone.get_default_timezone())
|
|
elif request.user.is_staff:
|
|
now = get_now()
|
|
else:
|
|
now = None
|
|
week, archive = models.Item.public(now)
|
|
context['items'] = week
|
|
if archive.exists():
|
|
context['archive'] = '/_%s/' % get_monday(get_now() - timedelta(days=7))
|
|
if now:
|
|
context['now'] = now
|
|
context['previous_week'] = (now - timedelta(days=7)).strftime(TS_FORMAT)
|
|
context['next_week'] = (now + timedelta(days=7)).strftime(TS_FORMAT)
|
|
return render(request, 'index.html', context)
|
|
|
|
|
|
def archive(request, year=None, month=None, day=None, week=None):
|
|
context = default_context(request)
|
|
archive = models.Item.all_public()
|
|
if year and month and day:
|
|
date = datetime.strptime('%s-%s-%s' % (year, month, day), '%Y-%m-%d')
|
|
week = int(date.strftime('%W'))
|
|
year = int(year)
|
|
archive_week = archive.filter(year=year, week=week).order_by('published')
|
|
years = {}
|
|
context['weeks'] = get_weeks(archive)
|
|
years["2024"] = []
|
|
for week in get_weeks(archive):
|
|
print(week)
|
|
if week["year"] not in years:
|
|
years[week["year"]] = []
|
|
years[week["year"]].append(week)
|
|
|
|
context['years'] = []
|
|
for year in reversed(sorted(years)):
|
|
context['years'].append({
|
|
'year': year,
|
|
'weeks': years[year]
|
|
})
|
|
context['this_week'] = date.strftime('%Y-%m-%d')
|
|
context['this_year'] = date.strftime('%Y')
|
|
extra = models.Week.objects.filter(monday=context['this_week']).first()
|
|
if extra:
|
|
context['week_title'] = extra.title
|
|
context['week_byline'] = extra.byline
|
|
elif week:
|
|
week = int(week)
|
|
year = int(year)
|
|
week = datetime.strptime('%s-W%s-1' % (year, week), "%Y-W%W-%w").strftime('%Y-%m-%d')
|
|
return redirect('/_%s/' % week)
|
|
elif year:
|
|
week = datetime.strptime('%s-W1-1' % year, "%Y-W%W-%w").strftime('%Y-%m-%d')
|
|
return redirect('/_%s/' % week)
|
|
context['items'] = archive_week
|
|
return render(request, 'archive.html', context)
|
|
|
|
|
|
def item(request, id):
|
|
context = default_context(request)
|
|
item = models.Item.objects.get(id=id)
|
|
if not request.user.is_staff and (
|
|
not item.published or item.published >= get_now()
|
|
):
|
|
raise Http404
|
|
context['item'] = item
|
|
context['url'] = request.build_absolute_uri(item.get_absolute_url())
|
|
qs = item.comments.order_by('created')
|
|
if not request.user.is_staff:
|
|
q = ~Q(published=None)
|
|
if request.user.is_authenticated:
|
|
q |= Q(user=request.user)
|
|
if request.session and request.session.session_key:
|
|
q |= Q(session_key=request.session.session_key)
|
|
qs = qs.filter(q)
|
|
comments = []
|
|
for comment in qs:
|
|
comments.append(comment.json())
|
|
context['comments'] = mark_safe(json.dumps(comments, ensure_ascii=False))
|
|
context['comments'] = json.dumps(comments, ensure_ascii=False)
|
|
|
|
user = {}
|
|
if request.user.is_staff:
|
|
user['is_moderator'] = True
|
|
if request.user.is_authenticated:
|
|
user['username'] = request.user.username
|
|
|
|
context['user'] = mark_safe(json.dumps(user))
|
|
request.session['item'] = id
|
|
qs = models.Item.objects.exclude(published=None).exclude(id=item.id)
|
|
if not request.user.is_staff:
|
|
now = get_now()
|
|
qs = qs.filter(published__lt=now)
|
|
previous_item = qs.exclude(published__gt=item.published).order_by('-published').first()
|
|
next_item = qs.exclude(published__lt=item.published).order_by('published').first()
|
|
if next_item:
|
|
context['next'] = next_item.get_absolute_url()
|
|
if previous_item:
|
|
context['previous'] = previous_item.get_absolute_url()
|
|
|
|
return render(request, 'item.html', context)
|
|
|
|
|
|
def comment(request):
|
|
response = {}
|
|
data = json.loads(request.body)
|
|
comment = models.Comment()
|
|
comment.item = models.Item.objects.get(id=data['item'])
|
|
if request.user.is_authenticated:
|
|
comment.user = request.user
|
|
if comment.user.has_perm('app.item.can_post_comment'):
|
|
comment.published = get_now()
|
|
else:
|
|
comment.name = data['name']
|
|
comment.email = data['email']
|
|
comment.session_key = request.session.session_key
|
|
comment.text = ox.add_links(data['text'])
|
|
comment.save()
|
|
link = request.build_absolute_uri(comment.item.get_absolute_url())
|
|
tasks.notify_moderators.delay(comment.id, link)
|
|
response = comment.json()
|
|
return render_to_json(response)
|
|
|
|
|
|
def publish_comment(request):
|
|
response = {}
|
|
data = json.loads(request.body)
|
|
if request.user.is_staff:
|
|
comment = models.Comment.objects.get(id=data['comment'])
|
|
comment.published = get_now()
|
|
comment.save()
|
|
if comment.data.get("moderator_ts"):
|
|
account = settings.SIGNAL_ACCOUNT
|
|
group = settings.SIGNAL_MODERATORS_GROUP
|
|
send_reaction(
|
|
account, comment.data["moderator_ts"], "🎉", group=group
|
|
)
|
|
link = request.build_absolute_uri(comment.item.get_absolute_url())
|
|
msg = '%s commented on <b>%s</b> at <a href="%s">%s</a>' % (
|
|
comment.name, comment.item.title, link, link
|
|
)
|
|
tasks.notify_telegram.delay(msg)
|
|
response['status'] = 'ok'
|
|
else:
|
|
response['error'] = 'permission denied'
|
|
return render_to_json(response)
|
|
|
|
|
|
def atom_xml(request):
|
|
feed = ET.Element("feed")
|
|
feed.attrib['xmlns'] = 'http://www.w3.org/2005/Atom'
|
|
feed.attrib['xmlns:media'] = 'http://search.yahoo.com/mrss/'
|
|
feed.attrib['xml:lang'] = 'en'
|
|
title = ET.SubElement(feed, "title")
|
|
title.text = settings.SITENAME
|
|
title.attrib['type'] = 'text'
|
|
link = ET.SubElement(feed, "link")
|
|
link.attrib['rel'] = 'self'
|
|
link.attrib['type'] = 'application/atom+xml'
|
|
atom_link = request.build_absolute_uri('/atom.xml')
|
|
link.attrib['href'] = atom_link
|
|
el = ET.SubElement(feed, 'id')
|
|
el.text = atom_link
|
|
|
|
week, archive = models.Item.public()
|
|
for item in week:
|
|
page_link = request.build_absolute_uri(item.get_absolute_url())
|
|
|
|
entry = ET.Element("entry")
|
|
title = ET.SubElement(entry, "title")
|
|
title.text = item.title
|
|
link = ET.SubElement(entry, "link")
|
|
link.attrib['rel'] = 'alternate'
|
|
link.attrib['href'] = page_link
|
|
updated = ET.SubElement(entry, "updated")
|
|
updated.text = item.modified.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
published = ET.SubElement(entry, "published")
|
|
published.text = item.created.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
el = ET.SubElement(entry, "id")
|
|
el.text = page_link
|
|
print(item.data)
|
|
if 'title' in item.data and 'thumbnail' in item.data:
|
|
html = f'''
|
|
<p>
|
|
{item.data['title']}
|
|
</p>
|
|
<img src={ item.data['thumbnail'] }>
|
|
'''.strip()
|
|
content = ET.SubElement(entry, "content")
|
|
content.attrib['type'] = 'html'
|
|
content.text = html
|
|
feed.append(entry)
|
|
return HttpResponse(
|
|
'<?xml version="1.0" encoding="utf-8" ?>\n' + ET.tostring(feed).decode(),
|
|
'application/atom+xml'
|
|
)
|
|
|