"""Pydantic models for Bluesky Dashboard."""
from datetime import datetime
from typing import Optional, List, Dict, Any, Literal
from pydantic import BaseModel, Field
# OAuth and Authentication
def to_camel(snake_str: str) -> str:
"""Convert snake_case to camelCase."""
components = snake_str.split("_")
return components[0] + "".join(x.title() for x in components[1:])
class OAuthConfig(BaseModel):
"""OAuth configuration response."""
public_origin: Optional[str] = None
is_configured: bool = False
allow_local_dev: bool = True
current_origin: str
class Config:
alias_generator = to_camel
populate_by_name = True
class BlueskySessionInfo(BaseModel):
"""Bluesky session information."""
did: str
handle: str
access_token: str
refresh_token: str
expires_at: datetime
# Feed Items and Authors
class FeedAuthor(BaseModel):
"""Author of a feed item."""
did: str
handle: str
display_name: str
avatar_url: Optional[str] = None
class FeedItem(BaseModel):
"""A feed item/post."""
subject_uri: str
actor_did: str
author_did: str
author_handle: str
author_display_name: str
text: str
origin: str
labels: Optional[List[str]] = None
created_at: Optional[datetime] = None
imported_at: datetime = Field(default_factory=datetime.utcnow)
class FeedItemViewModel(BaseModel):
"""Feed item with computed fields for display."""
subject_uri: str
author_did: str
author_handle: str
author_display_name: str
text: str
origin: str
labels: List[str] = Field(default_factory=list)
created_at: Optional[datetime] = None
score: float = 0.0
semantic_score: float = 0.0
author_affinity: float = 0.0
explanation: str = ""
embedding: Optional[List[float]] = None
class FeedAccountRecord(BaseModel):
"""Feed account information."""
did: str
handle: str
pds_url: Optional[str] = None
updated_at: datetime = Field(default_factory=datetime.utcnow)
# Feed Signals
class FeedSignal(BaseModel):
"""A user signal (like/dislike) on a feed item."""
subject_uri: str
item_key: str
generator_id: str
signal_type: Literal["like", "dislike", "neutral"]
created_at: datetime = Field(default_factory=datetime.utcnow)
text_preview: str = ""
explanation: str = ""
class FeedSignalRecord(BaseModel):
"""A stored feed signal with user context."""
signal: FeedSignal
actor_did: str
class StoredSignalRequest(BaseModel):
"""Request to add a signal."""
actor_did: str
signal: FeedSignal
# Feed State and Ranking
class LikeFeedState(BaseModel):
"""State of the like-based feed for a user."""
actor_did: Optional[str] = None
liked_authors: List[FeedAuthor] = Field(default_factory=list)
total_likes: int = 0
feed_items: List[FeedItem] = Field(default_factory=list)
class StoredFeedState(BaseModel):
"""Complete feed state including ranked items."""
actor_did: Optional[str] = None
generator_id: str
feed_items: List[FeedItemViewModel] = Field(default_factory=list)
last_updated: datetime = Field(default_factory=datetime.utcnow)
# Feed Generation
class GeneratorDefinition(BaseModel):
"""Definition of a feed generator."""
id: str
name: str
description: str
strategy: str = "hybrid"
icon: str = "star"
default_sort: Literal["chronological", "engagement", "ranked"] = "chronological"
default_time_range_hours: int = 168
class FeedGeneratorDescription(BaseModel):
"""Public description of a feed generator."""
feed_uri: str = ""
feed_key: str
name: str
description: str
is_configured: bool
service_did: Optional[str] = None
class PublishFeedRequest(BaseModel):
"""Request to publish a feed."""
actor_did: str
generator_id: str
feed_name: str
feed_description: str
is_public: bool = False
# Feed Skeleton (for Bluesky protocol)
class FeedSkeletonItem(BaseModel):
"""A single item in a feed skeleton."""
post: str = Field(..., description="AT URI of the post")
reason: Optional[Dict[str, Any]] = None
class FeedSkeletonResponse(BaseModel):
"""Feed skeleton response for getFeedSkeleton."""
feed: List[FeedSkeletonItem] = Field(default_factory=list)
cursor: Optional[str] = None
class DescribeFeedGeneratorFeed(BaseModel):
"""Feed info for describeFeedGenerator."""
uri: str
cid: str
class DescribeFeedGeneratorResponse(BaseModel):
"""Response for describeFeedGenerator XRPC method."""
did: str
feeds: List[DescribeFeedGeneratorFeed]
links: Dict[str, str] = Field(default_factory=dict)
# DID Document
class DidDocumentService(BaseModel):
"""Service entry in a DID document."""
id: str
type: str
service_endpoint: str
class DidDocument(BaseModel):
"""DID document for feed generator service."""
id: str = Field(..., alias="@id")
service: List[DidDocumentService] = Field(default_factory=list)
class Config:
populate_by_name = True
# Feed Import
class FeedImportPayload(BaseModel):
"""Payload for importing feed data from crawl."""
account: Dict[str, Any] # { did: str, handle: str, pdsUrl: str | null }
authors: List[Dict[str, Any]] # FeedAuthor[]
items: List[Dict[str, Any]] # FeedItem[]
seed_items: List[Dict[str, Any]] # FeedItem[]
class Config:
alias_generator = to_camel
populate_by_name = True
# Lists and Social
class ListMember(BaseModel):
"""A member of a list."""
did: str
handle: str
display_name: Optional[str] = None
class BlueskyList(BaseModel):
"""A Bluesky list."""
uri: str
name: str
description: Optional[str] = None
creator_did: str
members: List[ListMember] = Field(default_factory=list)
class BlueskyListCreator(BaseModel):
"""Information about a list creator."""
did: str
handle: str
list_count: int = 0
class FollowedPerson(BaseModel):
"""Someone the user follows."""
did: str
handle: str
display_name: Optional[str] = None
followed_at: Optional[datetime] = None
class FollowsAnalysisResult(BaseModel):
"""Result of analyzing follows."""
total_follows: int = 0
analyzed: int = 0
categorized: Dict[str, List[FollowedPerson]] = Field(default_factory=dict)
# Taste Profile
class TasteProfileContributor(BaseModel):
"""Something that contributes to taste profile."""
text: str
weight: float = 1.0
source: str = ""
class TasteProfileResult(BaseModel):
"""Computed taste profile for a user."""
actor_did: str
interests: List[str] = Field(default_factory=list)
keywords: List[str] = Field(default_factory=list)
sample_posts: List[FeedItemViewModel] = Field(default_factory=list)
generated_at: datetime = Field(default_factory=datetime.utcnow)