107 lines
2.7 KiB
Python
107 lines
2.7 KiB
Python
"""
|
|
子图数据模型和 Repository 定义
|
|
只需要定义表名、实体类，基类搞定一切
|
|
"""
|
|
from typing import List, Optional
|
|
from dataclasses import dataclass
|
|
|
|
from .base import BaseRepository, BaseEntity
|
|
|
|
|
|
# ========== 通讯录 ==========
|
|
|
|
@dataclass
|
|
class ContactEntity(BaseEntity):
|
|
"""通讯录实体"""
|
|
name: str = ""
|
|
phone: str = ""
|
|
email: str = ""
|
|
company: str = ""
|
|
position: str = ""
|
|
|
|
|
|
class ContactRepository(BaseRepository):
|
|
"""通讯录 Repository"""
|
|
table_name = "contacts"
|
|
entity_class = ContactEntity
|
|
|
|
# 如果需要特殊查询，在这里加方法
|
|
async def search_by_name(self, user_id: str, name_keyword: str) -> List[ContactEntity]:
|
|
"""自定义查询：按姓名搜索"""
|
|
sql = """
|
|
SELECT * FROM contacts
|
|
WHERE user_id = $1 AND name ILIKE $2
|
|
ORDER BY created_at DESC
|
|
LIMIT 50
|
|
"""
|
|
result = await self.custom_query(sql, (user_id, f"%{name_keyword}%"))
|
|
return [ContactEntity.from_dict(row) for row in result]
|
|
|
|
|
|
# ========== 词典 ==========
|
|
|
|
@dataclass
|
|
class WordEntity(BaseEntity):
|
|
"""单词实体"""
|
|
word: str = ""
|
|
phonetic: str = ""
|
|
part_of_speech: str = ""
|
|
definition: str = ""
|
|
examples: str = ""
|
|
|
|
|
|
class DictionaryRepository(BaseRepository):
|
|
"""词典 Repository"""
|
|
table_name = "words"
|
|
entity_class = WordEntity
|
|
|
|
async def search_by_word(self, user_id: str, word: str) -> Optional[WordEntity]:
|
|
"""按单词查询"""
|
|
sql = """
|
|
SELECT * FROM words
|
|
WHERE user_id = $1 AND word = $2
|
|
LIMIT 1
|
|
"""
|
|
result = await self.custom_query(sql, (user_id, word))
|
|
if result:
|
|
return WordEntity.from_dict(result[0])
|
|
return None
|
|
|
|
|
|
# ========== 资讯 ==========
|
|
|
|
@dataclass
|
|
class NewsEntity(BaseEntity):
|
|
"""资讯实体"""
|
|
title: str = ""
|
|
content: str = ""
|
|
url: str = ""
|
|
source: str = ""
|
|
keywords: str = ""
|
|
|
|
|
|
class NewsRepository(BaseRepository):
|
|
"""资讯 Repository"""
|
|
table_name = "news"
|
|
entity_class = NewsEntity
|
|
|
|
async def search_by_keywords(self, user_id: str, keyword: str) -> List[NewsEntity]:
|
|
"""按关键词搜索"""
|
|
sql = """
|
|
SELECT * FROM news
|
|
WHERE user_id = $1 AND (title ILIKE $2 OR keywords ILIKE $2)
|
|
ORDER BY created_at DESC
|
|
LIMIT 50
|
|
"""
|
|
result = await self.custom_query(sql, (user_id, f"%{keyword}%"))
|
|
return [NewsEntity.from_dict(row) for row in result]
|
|
|
|
|
|
# ========== 导出 ==========
|
|
|
|
__all__ = [
|
|
'ContactEntity', 'ContactRepository',
|
|
'WordEntity', 'DictionaryRepository',
|
|
'NewsEntity', 'NewsRepository',
|
|
]
|