398 lines
14 KiB
Python
398 lines
14 KiB
Python
import datetime
|
|
import os
|
|
|
|
from django.conf import settings
|
|
from django.contrib import messages
|
|
from django.contrib.auth.decorators import login_required
|
|
from django.core.exceptions import BadRequest
|
|
from django.db.models import Min
|
|
from django.http import HttpResponse
|
|
from django.shortcuts import redirect, render
|
|
from django.urls import reverse
|
|
from django.utils import timezone, translation
|
|
from django.utils.translation import gettext as _
|
|
|
|
from common.utils import GenerateDateUUIDMediaFilePath
|
|
from journal.exporters import CsvExporter, DoufenExporter, NdjsonExporter
|
|
from journal.importers import (
|
|
CsvImporter,
|
|
DoubanImporter,
|
|
GoodreadsImporter,
|
|
LetterboxdImporter,
|
|
NdjsonImporter,
|
|
OPMLImporter,
|
|
)
|
|
from journal.models import ShelfType
|
|
from takahe.utils import Takahe
|
|
from users.models import Task
|
|
|
|
from .account import clear_preference_cache
|
|
|
|
|
|
@login_required
|
|
def preferences(request):
|
|
if not request.user.registration_complete:
|
|
return redirect(reverse("users:register"))
|
|
preference = request.user.preference
|
|
identity = request.user.identity
|
|
if request.method == "POST":
|
|
identity.anonymous_viewable = bool(request.POST.get("anonymous_viewable"))
|
|
identity.save(update_fields=["anonymous_viewable"])
|
|
tidentity = Takahe.get_identity(identity.pk)
|
|
tidentity.indexable = bool(request.POST.get("anonymous_viewable"))
|
|
tidentity.save(update_fields=["indexable"])
|
|
|
|
preference.default_visibility = int(request.POST.get("default_visibility"))
|
|
preference.mastodon_default_repost = (
|
|
int(request.POST.get("mastodon_default_repost", 0)) == 1
|
|
)
|
|
preference.classic_homepage = int(request.POST.get("classic_homepage"))
|
|
preference.hidden_categories = request.POST.getlist("hidden_categories")
|
|
preference.post_public_mode = int(request.POST.get("post_public_mode"))
|
|
preference.show_last_edit = bool(request.POST.get("show_last_edit"))
|
|
preference.mastodon_repost_mode = int(
|
|
request.POST.get("mastodon_repost_mode", 0)
|
|
)
|
|
preference.mastodon_append_tag = request.POST.get(
|
|
"mastodon_append_tag", ""
|
|
).strip()
|
|
preference.save(
|
|
update_fields=[
|
|
"default_visibility",
|
|
"post_public_mode",
|
|
"classic_homepage",
|
|
"mastodon_append_tag",
|
|
"mastodon_repost_mode",
|
|
"mastodon_default_repost",
|
|
"show_last_edit",
|
|
"hidden_categories",
|
|
]
|
|
)
|
|
lang = request.POST.get("language")
|
|
if lang in dict(settings.LANGUAGES).keys() and lang != request.user.language:
|
|
request.user.language = lang
|
|
translation.activate(lang)
|
|
request.LANGUAGE_CODE = translation.get_language()
|
|
request.user.save(update_fields=["language"])
|
|
clear_preference_cache(request)
|
|
return render(
|
|
request,
|
|
"users/preferences.html",
|
|
{"enable_local_only": settings.ENABLE_LOCAL_ONLY},
|
|
)
|
|
|
|
|
|
@login_required
|
|
def data(request):
|
|
if not request.user.registration_complete:
|
|
return redirect(reverse("users:register"))
|
|
current_year = datetime.date.today().year
|
|
queryset = request.user.identity.shelf_manager.get_shelf(
|
|
ShelfType.COMPLETE
|
|
).members.all()
|
|
start_date = queryset.aggregate(Min("created_time"))["created_time__min"]
|
|
start_year = start_date.year if start_date else current_year
|
|
years = reversed(range(start_year, current_year + 1))
|
|
|
|
# Import tasks - check for both CSV and NDJSON importers
|
|
csv_import_task = CsvImporter.latest_task(request.user)
|
|
ndjson_import_task = NdjsonImporter.latest_task(request.user)
|
|
# Use the most recent import task for display
|
|
if ndjson_import_task and (
|
|
not csv_import_task
|
|
or ndjson_import_task.created_time > csv_import_task.created_time
|
|
):
|
|
neodb_import_task = ndjson_import_task
|
|
else:
|
|
neodb_import_task = csv_import_task
|
|
|
|
return render(
|
|
request,
|
|
"users/data.html",
|
|
{
|
|
"allow_any_site": settings.MASTODON_ALLOW_ANY_SITE,
|
|
"import_task": DoubanImporter.latest_task(request.user),
|
|
"export_task": DoufenExporter.latest_task(request.user),
|
|
"csv_export_task": CsvExporter.latest_task(request.user),
|
|
"neodb_import_task": neodb_import_task, # Use the most recent import task
|
|
"ndjson_export_task": NdjsonExporter.latest_task(request.user),
|
|
"letterboxd_task": LetterboxdImporter.latest_task(request.user),
|
|
"goodreads_task": GoodreadsImporter.latest_task(request.user),
|
|
# "opml_task": OPMLImporter.latest_task(request.user),
|
|
"years": years,
|
|
},
|
|
)
|
|
|
|
|
|
@login_required
|
|
def data_import_status(request):
|
|
return render(
|
|
request,
|
|
"users/data_import_status.html",
|
|
{
|
|
"import_task": DoubanImporter.latest_task(request.user),
|
|
},
|
|
)
|
|
|
|
|
|
@login_required
|
|
def user_task_status(request, task_type: str):
|
|
match task_type:
|
|
case "journal.csvimporter":
|
|
task_cls = CsvImporter
|
|
case "journal.ndjsonimporter":
|
|
task_cls = NdjsonImporter
|
|
case "journal.csvexporter":
|
|
task_cls = CsvExporter
|
|
case "journal.ndjsonexporter":
|
|
task_cls = NdjsonExporter
|
|
case "journal.letterboxdimporter":
|
|
task_cls = LetterboxdImporter
|
|
case "journal.goodreadsimporter":
|
|
task_cls = GoodreadsImporter
|
|
case "journal.opmlimporter":
|
|
task_cls = OPMLImporter
|
|
case "journal.doubanimporter":
|
|
task_cls = DoubanImporter
|
|
case _:
|
|
return redirect(reverse("users:data"))
|
|
task = task_cls.latest_task(request.user)
|
|
return render(request, "users/user_task_status.html", {"task": task})
|
|
|
|
|
|
@login_required
|
|
def user_task_download(request, task_type: str):
|
|
match task_type:
|
|
case "journal.csvexporter":
|
|
task_cls = CsvExporter
|
|
case "journal.ndjsonexporter":
|
|
task_cls = NdjsonExporter
|
|
case _:
|
|
return redirect(reverse("users:data"))
|
|
task = task_cls.latest_task(request.user)
|
|
if not task or task.state != Task.States.complete or not task.metadata.get("file"):
|
|
messages.add_message(request, messages.ERROR, _("Export file not available."))
|
|
return redirect(reverse("users:data"))
|
|
response = HttpResponse()
|
|
response["X-Accel-Redirect"] = (
|
|
settings.MEDIA_URL + task.metadata["file"][len(settings.MEDIA_ROOT) :]
|
|
)
|
|
response["Content-Type"] = "application/zip"
|
|
response["Content-Disposition"] = f'attachment; filename="{task.filename}.zip"'
|
|
return response
|
|
|
|
|
|
@login_required
|
|
def export_reviews(request):
|
|
if request.method != "POST":
|
|
return redirect(reverse("users:data"))
|
|
return render(request, "users/data.html")
|
|
|
|
|
|
@login_required
|
|
def export_marks(request):
|
|
# TODO: deprecated
|
|
if request.method == "POST":
|
|
DoufenExporter.create(request.user).enqueue()
|
|
messages.add_message(request, messages.INFO, _("Generating exports."))
|
|
return redirect(reverse("users:data"))
|
|
else:
|
|
task = DoufenExporter.latest_task(request.user)
|
|
if not task or task.state != Task.States.complete:
|
|
messages.add_message(
|
|
request, messages.ERROR, _("Export file not available.")
|
|
)
|
|
return redirect(reverse("users:data"))
|
|
try:
|
|
with open(task.metadata["file"], "rb") as fh:
|
|
response = HttpResponse(
|
|
fh.read(), content_type="application/vnd.ms-excel"
|
|
)
|
|
response["Content-Disposition"] = 'attachment;filename="marks.xlsx"'
|
|
return response
|
|
except Exception:
|
|
messages.add_message(
|
|
request, messages.ERROR, _("Export file expired. Please export again.")
|
|
)
|
|
return redirect(reverse("users:data"))
|
|
|
|
|
|
@login_required
|
|
def export_csv(request):
|
|
if request.method == "POST":
|
|
task = CsvExporter.latest_task(request.user)
|
|
if (
|
|
task
|
|
and task.state not in [Task.States.complete, Task.States.failed]
|
|
and task.created_time > (timezone.now() - datetime.timedelta(hours=1))
|
|
):
|
|
messages.add_message(
|
|
request, messages.INFO, _("Recent export still in progress.")
|
|
)
|
|
return redirect(reverse("users:data"))
|
|
CsvExporter.create(request.user).enqueue()
|
|
return redirect(
|
|
reverse("users:user_task_status", args=("journal.csvexporter",))
|
|
)
|
|
return redirect(reverse("users:data"))
|
|
|
|
|
|
@login_required
|
|
def export_ndjson(request):
|
|
if request.method == "POST":
|
|
task = NdjsonExporter.latest_task(request.user)
|
|
if (
|
|
task
|
|
and task.state not in [Task.States.complete, Task.States.failed]
|
|
and task.created_time > (timezone.now() - datetime.timedelta(hours=1))
|
|
):
|
|
messages.add_message(
|
|
request, messages.INFO, _("Recent export still in progress.")
|
|
)
|
|
return redirect(reverse("users:data"))
|
|
NdjsonExporter.create(request.user).enqueue()
|
|
return redirect(
|
|
reverse("users:user_task_status", args=("journal.ndjsonexporter",))
|
|
)
|
|
return redirect(reverse("users:data"))
|
|
|
|
|
|
@login_required
|
|
def sync_mastodon(request):
|
|
if request.method == "POST":
|
|
request.user.sync_accounts_later()
|
|
messages.add_message(request, messages.INFO, _("Sync in progress."))
|
|
return redirect(reverse("users:info"))
|
|
|
|
|
|
@login_required
|
|
def sync_mastodon_preference(request):
|
|
if request.method == "POST":
|
|
request.user.preference.mastodon_skip_userinfo = (
|
|
request.POST.get("mastodon_sync_userinfo", "") == ""
|
|
)
|
|
request.user.preference.mastodon_skip_relationship = (
|
|
request.POST.get("mastodon_sync_relationship", "") == ""
|
|
)
|
|
request.user.preference.save()
|
|
messages.add_message(request, messages.INFO, _("Settings saved."))
|
|
return redirect(reverse("users:info"))
|
|
|
|
|
|
@login_required
|
|
def import_goodreads(request):
|
|
if request.method != "POST":
|
|
return redirect(reverse("users:data"))
|
|
raw_url = request.POST.get("url")
|
|
if not GoodreadsImporter.validate_url(raw_url):
|
|
raise BadRequest(_("Invalid URL."))
|
|
task = GoodreadsImporter.create(
|
|
request.user,
|
|
visibility=int(request.POST.get("visibility", 0)),
|
|
url=raw_url,
|
|
)
|
|
task.enqueue()
|
|
return redirect(reverse("users:user_task_status", args=(task.type,)))
|
|
|
|
|
|
@login_required
|
|
def import_douban(request):
|
|
if request.method != "POST":
|
|
return redirect(reverse("users:data"))
|
|
if not DoubanImporter.validate_file(request.FILES["file"]):
|
|
raise BadRequest(_("Invalid file."))
|
|
f = (
|
|
settings.MEDIA_ROOT
|
|
+ "/"
|
|
+ GenerateDateUUIDMediaFilePath("x.zip", settings.SYNC_FILE_PATH_ROOT)
|
|
)
|
|
os.makedirs(os.path.dirname(f), exist_ok=True)
|
|
with open(f, "wb+") as destination:
|
|
for chunk in request.FILES["file"].chunks():
|
|
destination.write(chunk)
|
|
task = DoubanImporter.create(
|
|
request.user,
|
|
visibility=int(request.POST.get("visibility", 0)),
|
|
mode=int(request.POST.get("import_mode", 0)),
|
|
file=f,
|
|
)
|
|
task.enqueue()
|
|
return redirect(reverse("users:user_task_status", args=(task.type,)))
|
|
|
|
|
|
@login_required
|
|
def import_letterboxd(request):
|
|
if request.method != "POST":
|
|
return redirect(reverse("users:data"))
|
|
if not LetterboxdImporter.validate_file(request.FILES["file"]):
|
|
raise BadRequest(_("Invalid file."))
|
|
f = (
|
|
settings.MEDIA_ROOT
|
|
+ "/"
|
|
+ GenerateDateUUIDMediaFilePath("x.zip", settings.SYNC_FILE_PATH_ROOT)
|
|
)
|
|
os.makedirs(os.path.dirname(f), exist_ok=True)
|
|
with open(f, "wb+") as destination:
|
|
for chunk in request.FILES["file"].chunks():
|
|
destination.write(chunk)
|
|
task = LetterboxdImporter.create(
|
|
request.user,
|
|
visibility=int(request.POST.get("visibility", 0)),
|
|
file=f,
|
|
)
|
|
task.enqueue()
|
|
return redirect(reverse("users:user_task_status", args=(task.type,)))
|
|
|
|
|
|
@login_required
|
|
def import_opml(request):
|
|
if request.method != "POST":
|
|
return redirect(reverse("users:data"))
|
|
if not OPMLImporter.validate_file(request.FILES["file"]):
|
|
raise BadRequest(_("Invalid file."))
|
|
f = (
|
|
settings.MEDIA_ROOT
|
|
+ "/"
|
|
+ GenerateDateUUIDMediaFilePath("x.zip", settings.SYNC_FILE_PATH_ROOT)
|
|
)
|
|
os.makedirs(os.path.dirname(f), exist_ok=True)
|
|
with open(f, "wb+") as destination:
|
|
for chunk in request.FILES["file"].chunks():
|
|
destination.write(chunk)
|
|
task = OPMLImporter.create(
|
|
request.user,
|
|
visibility=int(request.POST.get("visibility", 0)),
|
|
mode=int(request.POST.get("import_mode", 0)),
|
|
file=f,
|
|
)
|
|
task.enqueue()
|
|
return redirect(reverse("users:user_task_status", args=(task.type,)))
|
|
|
|
|
|
@login_required
|
|
def import_neodb(request):
|
|
if request.method == "POST":
|
|
format_type_hint = request.POST.get("format_type", "").lower()
|
|
if format_type_hint == "csv":
|
|
importer = CsvImporter
|
|
elif format_type_hint == "ndjson":
|
|
importer = NdjsonImporter
|
|
else:
|
|
raise BadRequest("Invalid file.")
|
|
f = (
|
|
settings.MEDIA_ROOT
|
|
+ "/"
|
|
+ GenerateDateUUIDMediaFilePath("x.zip", settings.SYNC_FILE_PATH_ROOT)
|
|
)
|
|
os.makedirs(os.path.dirname(f), exist_ok=True)
|
|
with open(f, "wb+") as destination:
|
|
for chunk in request.FILES["file"].chunks():
|
|
destination.write(chunk)
|
|
task = importer.create(
|
|
request.user,
|
|
visibility=int(request.POST.get("visibility", 0)),
|
|
file=f,
|
|
)
|
|
task.enqueue()
|
|
return redirect(reverse("users:user_task_status", args=(task.type,)))
|
|
return redirect(reverse("users:data"))
|