models.py•4.48 kB
from datetime import datetime
from typing import Optional, List, Dict, Any
from sqlmodel import SQLModel, Field, JSON, Column
from sqlalchemy import DateTime
import json
class User(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
phone: str = Field(unique=True, index=True)
created_at: datetime = Field(default_factory=datetime.utcnow)
class TrackedQuery(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
user_phone: str = Field(foreign_key="user.phone", index=True)
query_text: str = Field(index=True)
competitors_json: str = Field(default="[]") # JSON string of competitors list
created_at: datetime = Field(default_factory=datetime.utcnow)
@property
def competitors(self) -> List[str]:
return json.loads(self.competitors_json)
@competitors.setter
def competitors(self, value: List[str]):
self.competitors_json = json.dumps(value)
class Run(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
user_phone: str = Field(foreign_key="user.phone", index=True)
query_id: int = Field(foreign_key="trackedquery.id", index=True)
run_at: datetime = Field(default_factory=datetime.utcnow)
platforms_summary_json: str = Field(default="{}") # JSON string of platform status
@property
def platforms_summary(self) -> Dict[str, Any]:
return json.loads(self.platforms_summary_json)
@platforms_summary.setter
def platforms_summary(self, value: Dict[str, Any]):
self.platforms_summary_json = json.dumps(value)
class RunItem(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
run_id: int = Field(foreign_key="run.id", index=True)
platform: str = Field(index=True)
raw_answer: str
mentions_json: str = Field(default="[]") # JSON string of mentions
citations_json: str = Field(default="[]") # JSON string of citations
sentiment_json: str = Field(default="{}") # JSON string of sentiment
first_position_brand: Optional[str] = Field(default=None)
stats_json: str = Field(default="{}") # JSON string of stats (tokens, latency)
created_at: datetime = Field(default_factory=datetime.utcnow)
@property
def mentions(self) -> List[Dict[str, Any]]:
return json.loads(self.mentions_json)
@mentions.setter
def mentions(self, value: List[Dict[str, Any]]):
self.mentions_json = json.dumps(value)
@property
def citations(self) -> List[Dict[str, Any]]:
return json.loads(self.citations_json)
@citations.setter
def citations(self, value: List[Dict[str, Any]]):
self.citations_json = json.dumps(value)
@property
def sentiment(self) -> Dict[str, Any]:
return json.loads(self.sentiment_json)
@sentiment.setter
def sentiment(self, value: Dict[str, Any]):
self.sentiment_json = json.dumps(value)
@property
def stats(self) -> Dict[str, Any]:
return json.loads(self.stats_json)
@stats.setter
def stats(self, value: Dict[str, Any]):
self.stats_json = json.dumps(value)
class Aggregate(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
user_phone: str = Field(foreign_key="user.phone", index=True)
query_id: int = Field(foreign_key="trackedquery.id", index=True)
day: str = Field(index=True) # YYYY-MM-DD format
sov_json: str = Field(default="[]") # JSON string of share of voice
sentiment_json: str = Field(default="[]") # JSON string of sentiment
citations_json: str = Field(default="[]") # JSON string of citations
updated_at: datetime = Field(default_factory=datetime.utcnow)
@property
def sov(self) -> List[Dict[str, Any]]:
return json.loads(self.sov_json)
@sov.setter
def sov(self, value: List[Dict[str, Any]]):
self.sov_json = json.dumps(value)
@property
def sentiment(self) -> List[Dict[str, Any]]:
return json.loads(self.sentiment_json)
@sentiment.setter
def sentiment(self, value: List[Dict[str, Any]]):
self.sentiment_json = json.dumps(value)
@property
def citations(self) -> List[Dict[str, Any]]:
return json.loads(self.citations_json)
@citations.setter
def citations(self, value: List[Dict[str, Any]]):
self.citations_json = json.dumps(value)