Spaces:
Sleeping
Sleeping
File size: 3,549 Bytes
5e0532d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 |
from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum
import sqlite3
import json
class TonePreference(str, Enum):
PASTORAL = "pastoral"
SCHOLARLY = "scholarly"
GENTLE = "gentle"
DIRECT = "direct"
class SpiritualGoal(str, Enum):
PRAYER = "prayer"
STUDY = "study"
HEALING = "healing"
LEADERSHIP = "leadership"
DISCERNMENT = "discernment"
class UserProfile(BaseModel):
user_id: str
denomination: Optional[str] = None
bible_translations: List[str] = Field(default=["NIV", "ESV"])
tone_preference: TonePreference = TonePreference.PASTORAL
spiritual_goals: List[SpiritualGoal] = []
class Config:
json_schema_extra = {
"example": {
"user_id": "usr_123",
"denomination": "non-denominational",
"bible_translations": ["NIV"],
"tone_preference": "gentle",
"spiritual_goals": ["prayer", "healing"]
}
}
class ProfileService:
def __init__(self, db_path: str = "memory.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS user_profiles (
user_id TEXT PRIMARY KEY,
denomination TEXT,
bible_translations TEXT,
tone_preference TEXT,
spiritual_goals TEXT
)
''')
conn.commit()
conn.close()
async def get_profile(self, user_id: str) -> UserProfile:
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()
cursor.execute('SELECT * FROM user_profiles WHERE user_id = ?', (user_id,))
row = cursor.fetchone()
conn.close()
if row:
return UserProfile(
user_id=row['user_id'],
denomination=row['denomination'],
bible_translations=json.loads(row['bible_translations']) if row['bible_translations'] else ["NIV"],
tone_preference=row['tone_preference'] or TonePreference.PASTORAL,
spiritual_goals=json.loads(row['spiritual_goals']) if row['spiritual_goals'] else []
)
# Return default profile if none exists
return UserProfile(user_id=user_id)
async def update_profile(self, user_id: str, data: dict) -> UserProfile:
# Get current profile to merge (or default)
current = await self.get_profile(user_id)
# Merge data and validate to ensure Enums are correctly parsed
merged_data = current.model_dump()
merged_data.update(data)
updated_model = UserProfile(**merged_data)
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
INSERT OR REPLACE INTO user_profiles (user_id, denomination, bible_translations, tone_preference, spiritual_goals)
VALUES (?, ?, ?, ?, ?)
''', (
updated_model.user_id,
updated_model.denomination,
json.dumps(updated_model.bible_translations),
updated_model.tone_preference.value,
json.dumps([g.value for g in updated_model.spiritual_goals])
))
conn.commit()
conn.close()
return updated_model
profile_service = ProfileService()
|