mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
248 lines
9.0 KiB
Python
248 lines
9.0 KiB
Python
from rest_framework import serializers
|
|
from .models import Link, Page, Post, ImageCollection, Image, Tag, FileUpload, Bookmark
|
|
|
|
|
|
class LinkSerializer(serializers.ModelSerializer):
|
|
tags = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = Link
|
|
fields = ['id', 'alias', 'original_url', 'description', 'link_type',
|
|
'click_count', 'tags', 'created_at', 'updated_at']
|
|
read_only_fields = ['id', 'click_count', 'created_at', 'updated_at']
|
|
|
|
def get_tags(self, obj):
|
|
return [{'id': tag.id, 'name': tag.name, 'slug': tag.slug} for tag in obj.tags.all()]
|
|
|
|
class PageSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = Page
|
|
fields = ['id', 'url', 'title', 'summary', 'content', 'screenshot_path',
|
|
'process_status', 'created_at', 'updated_at']
|
|
read_only_fields = ['id', 'screenshot_path', 'process_status',
|
|
'created_at', 'updated_at']
|
|
|
|
class PostSerializer(serializers.ModelSerializer):
|
|
tags = serializers.ListField(child=serializers.CharField(), required=False, write_only=True, help_text="List of tag slugs")
|
|
tag_details = serializers.SerializerMethodField(read_only=True)
|
|
|
|
class Meta:
|
|
model = Post
|
|
fields = ['id', 'title', 'summary', 'content', 'is_public', 'tags', 'tag_details', 'created_at', 'updated_at']
|
|
read_only_fields = ['id', 'created_at', 'updated_at']
|
|
extra_kwargs = {
|
|
'title': {'required': True},
|
|
'content': {'required': True},
|
|
'summary': {'required': False}
|
|
}
|
|
|
|
def get_tag_details(self, obj):
|
|
return [{'id': tag.id, 'name': tag.name, 'slug': tag.slug} for tag in obj.tags.all()]
|
|
|
|
def create(self, validated_data):
|
|
tag_slugs = validated_data.pop('tags', [])
|
|
post = Post.objects.create(**validated_data)
|
|
|
|
for tag_slug in tag_slugs:
|
|
try:
|
|
tag = Tag.objects.get(slug=tag_slug)
|
|
except Tag.DoesNotExist:
|
|
# If tag doesn't exist, create it with the slug as both name and slug
|
|
tag = Tag.objects.create(name=tag_slug, slug=tag_slug)
|
|
post.tags.add(tag)
|
|
|
|
return post
|
|
|
|
def update(self, instance, validated_data):
|
|
tag_slugs = validated_data.pop('tags', None)
|
|
for attr, value in validated_data.items():
|
|
setattr(instance, attr, value)
|
|
|
|
if tag_slugs is not None:
|
|
instance.tags.clear()
|
|
for tag_slug in tag_slugs:
|
|
try:
|
|
tag = Tag.objects.get(slug=tag_slug)
|
|
except Tag.DoesNotExist:
|
|
# If tag doesn't exist, create it with the slug as both name and slug
|
|
tag = Tag.objects.create(name=tag_slug, slug=tag_slug)
|
|
instance.tags.add(tag)
|
|
|
|
instance.save()
|
|
return instance
|
|
|
|
class ImageSerializer(serializers.ModelSerializer):
|
|
url = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = Image
|
|
fields = ['id', 'collection', 'title', 'description', 'content_type', 'size', 'created_at', 'updated_at', 'url']
|
|
read_only_fields = ['id', 'collection', 'content_type', 'size', 'created_at', 'updated_at']
|
|
|
|
def get_url(self, obj):
|
|
return obj.get_url()
|
|
|
|
class ImageDescriptionSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = Image
|
|
fields = ['description']
|
|
|
|
class ImageCollectionSerializer(serializers.ModelSerializer):
|
|
image_count = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = ImageCollection
|
|
fields = ['id', 'name', 'description', 'image_count', 'created_at']
|
|
|
|
def get_image_count(self, obj):
|
|
return obj.images.count()
|
|
|
|
class FileUploadSerializer(serializers.ModelSerializer):
|
|
formatted_size = serializers.SerializerMethodField()
|
|
public_url = serializers.SerializerMethodField()
|
|
is_expired = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = FileUpload
|
|
fields = [
|
|
'id', 'name', 'mime_type', 'size', 'formatted_size',
|
|
'is_public', 'public_token', 'public_url',
|
|
'expires_at', 'is_expired',
|
|
'download_count', 'created_at', 'updated_at',
|
|
]
|
|
read_only_fields = [
|
|
'id', 'mime_type', 'size', 'formatted_size',
|
|
'public_token', 'public_url', 'is_expired',
|
|
'download_count', 'created_at', 'updated_at',
|
|
]
|
|
|
|
def get_formatted_size(self, obj):
|
|
return obj.formatted_size()
|
|
|
|
def get_public_url(self, obj):
|
|
return obj.public_url
|
|
|
|
def get_is_expired(self, obj):
|
|
return obj.is_expired
|
|
|
|
|
|
class BookmarkSerializer(serializers.ModelSerializer):
|
|
"""Full bookmark for read / single-create / update."""
|
|
|
|
class Meta:
|
|
model = Bookmark
|
|
fields = [
|
|
'id', 'tweet_id', 'url',
|
|
'author_screen_name', 'author_name', 'author_profile_image_url',
|
|
'text', 'summary',
|
|
'tweet_created_at', 'bookmark_created_at',
|
|
'view_count', 'favorite_count', 'retweet_count', 'reply_count',
|
|
'bookmark_count', 'quote_count',
|
|
'has_video', 'has_photo', 'has_card', 'is_quote', 'is_retweet',
|
|
'media', 'card_data', 'quoted_data',
|
|
'data',
|
|
'created_at', 'updated_at',
|
|
]
|
|
read_only_fields = ['id', 'created_at', 'updated_at']
|
|
|
|
|
|
def _norm_bookmark_for_import(raw, obj=None):
|
|
"""
|
|
Map a normalised x-bookmarks-exporter object (or any compatible dict) onto
|
|
a Bookmark instance so it can be upserted. Reused by API import + mgmt cmd.
|
|
"""
|
|
author = raw.get('author') or {}
|
|
media = raw.get('media') or []
|
|
|
|
tweet_id = raw.get('id') or raw.get('tweet_id')
|
|
if not tweet_id:
|
|
raise serializers.ValidationError('missing id/tweet_id')
|
|
|
|
b = obj or Bookmark(tweet_id=tweet_id)
|
|
b.tweet_id = str(tweet_id)
|
|
b.url = raw.get('url') or ''
|
|
b.author_screen_name = author.get('screen_name') or ''
|
|
b.author_name = author.get('name') or ''
|
|
b.author_profile_image_url = author.get('profile_image_url') or ''
|
|
b.text = raw.get('text') or ''
|
|
b.summary = raw.get('summary') or ''
|
|
|
|
# parse tweet created_at ("Wed Jul 30 12:00:00 +0000 2025" style) — leave
|
|
# None if not parseable, view layer handles null.
|
|
from datetime import datetime
|
|
tca = raw.get('created_at')
|
|
if isinstance(tca, str) and tca:
|
|
try:
|
|
b.tweet_created_at = datetime.strptime(tca, '%a %b %d %H:%M:%S %z %Y')
|
|
except ValueError:
|
|
b.tweet_created_at = None
|
|
elif isinstance(tca, datetime):
|
|
b.tweet_created_at = tca
|
|
|
|
bca = raw.get('bookmark_created_at')
|
|
if isinstance(bca, str) and bca:
|
|
try:
|
|
b.bookmark_created_at = datetime.strptime(
|
|
bca, '%a %b %d %H:%M:%S %z %Y')
|
|
except ValueError:
|
|
b.bookmark_created_at = None
|
|
elif isinstance(bca, datetime):
|
|
b.bookmark_created_at = bca
|
|
|
|
b.view_count = raw.get('view_count') or 0
|
|
b.favorite_count = raw.get('favorite_count') or 0
|
|
b.retweet_count = raw.get('retweet_count') or 0
|
|
b.reply_count = raw.get('reply_count') or 0
|
|
b.bookmark_count = raw.get('bookmark_count') or 0
|
|
b.quote_count = raw.get('quote_count') or 0
|
|
|
|
b.has_video = any(m.get('type') in ('video', 'animated_gif') for m in media)
|
|
b.has_photo = any(m.get('type') == 'photo' for m in media)
|
|
b.has_card = bool(raw.get('card'))
|
|
b.is_quote = bool(raw.get('quoted'))
|
|
b.is_retweet = bool(raw.get('retweet'))
|
|
|
|
b.media = media
|
|
b.card_data = raw.get('card')
|
|
b.quoted_data = raw.get('quoted')
|
|
b.data = raw
|
|
return b
|
|
|
|
|
|
class BookmarkImportItemSerializer(serializers.Serializer):
|
|
"""One element of a bulk-import payload. Accepts the full export object."""
|
|
def to_internal_value(self, data):
|
|
# we don't validate strictly — _norm_bookmark_for_import does the mapping
|
|
return data
|
|
|
|
def create(self, validated):
|
|
tweet_id = str(validated.get('id') or validated.get('tweet_id'))
|
|
obj = Bookmark.objects.filter(tweet_id=tweet_id).first()
|
|
b = _norm_bookmark_for_import(validated, obj)
|
|
b.save()
|
|
return b
|
|
|
|
def update(self, instance, validated):
|
|
return _norm_bookmark_for_import(validated, instance).save() or instance
|
|
|
|
|
|
class BookmarkImportSerializer(serializers.Serializer):
|
|
"""Top-level bulk-import payload: {bookmarks: [ ... ]}."""
|
|
bookmarks = BookmarkImportItemSerializer(many=True)
|
|
|
|
def create(self, validated):
|
|
items = validated.get('bookmarks', [])
|
|
created = updated = skipped = 0
|
|
for raw in items:
|
|
tweet_id = str(raw.get('id') or raw.get('tweet_id') or '')
|
|
if not tweet_id:
|
|
skipped += 1
|
|
continue
|
|
existed = Bookmark.objects.filter(tweet_id=tweet_id).exists()
|
|
BookmarkImportItemSerializer().create(raw)
|
|
if existed:
|
|
updated += 1
|
|
else:
|
|
created += 1
|
|
return {'created': created, 'updated': updated, 'skipped': skipped}
|