Files
ewoooc/database/ai_models.py
ogt 1b4f3a7bbe
Some checks failed
CD Pipeline / deploy (push) Failing after 59s
feat: EwoooC 初始化 — 完整專案推版至 Gitea
- 建立 Gitea Actions CD pipeline (.gitea/workflows/cd.yaml)
- 部署模式: rsync Python 檔案至 188 → docker restart (volume mount)
- Dockerfile/requirements 變動時自動重建 Docker image
- 部署通知: Telegram (開始/成功/失敗)
- 健康檢查: https://mo.wooo.work/health (最多 5 次重試)
- 同步最新 CLAUDE.md / ADR-008 / memory (2026-04-19)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-19 01:21:13 +08:00

320 lines
12 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
AI 生成歷史記錄資料庫模型
儲存 Ollama/Gemini LLM 生成的文案和推薦結果
支援多 AI 提供者和費用追蹤
"""
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Text, Boolean, ForeignKey, Index
from sqlalchemy.orm import relationship
from datetime import datetime, date
from .models import Base
class AIGenerationHistory(Base):
"""AI 生成歷史記錄表"""
__tablename__ = 'ai_generation_history'
id = Column(Integer, primary_key=True)
# 生成類型copy (文案), recommend (推薦), weather_analysis (天氣分析)
generation_type = Column(String(50), nullable=False, index=True)
# 商品相關
product_name = Column(String(255), index=True)
# 輸入參數
input_keywords = Column(Text) # JSON 格式的關鍵字列表
input_style = Column(String(50)) # 文案風格
input_trend_topic = Column(Text) # 趨勢話題(用於推薦)
# 生成結果
output_content = Column(Text, nullable=False) # 生成的內容
# 模型資訊
model_name = Column(String(100))
generation_duration = Column(Float) # 生成耗時(秒)
# AI 提供者資訊 (新增 - 支援 Ollama/Gemini 切換)
ai_provider = Column(String(20), default='ollama') # 'ollama' 或 'gemini'
input_tokens = Column(Integer, default=0) # 輸入 token 數量 (用於 Gemini 費用計算)
output_tokens = Column(Integer, default=0) # 輸出 token 數量
# 評價與狀態
rating = Column(Integer) # 用戶評分 1-5
is_favorite = Column(Boolean, default=False) # 是否收藏
is_used = Column(Boolean, default=False) # 是否已使用
notes = Column(Text) # 用戶備註
# 用戶追蹤
created_by = Column(Integer, ForeignKey('users.id'))
created_at = Column(DateTime, default=datetime.now, index=True)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
# 建立索引以優化查詢
__table_args__ = (
Index('idx_ai_history_type_created', 'generation_type', 'created_at'),
Index('idx_ai_history_product', 'product_name'),
Index('idx_ai_history_favorite', 'is_favorite', 'created_at'),
)
def to_dict(self):
"""轉換為字典格式"""
import json
return {
'id': self.id,
'generation_type': self.generation_type,
'product_name': self.product_name,
'input_keywords': json.loads(self.input_keywords) if self.input_keywords else [],
'input_style': self.input_style,
'input_trend_topic': self.input_trend_topic,
'output_content': self.output_content,
'model_name': self.model_name,
'generation_duration': self.generation_duration,
'ai_provider': self.ai_provider,
'input_tokens': self.input_tokens,
'output_tokens': self.output_tokens,
'rating': self.rating,
'is_favorite': self.is_favorite,
'is_used': self.is_used,
'notes': self.notes,
'created_by': self.created_by,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
}
class AIUsageTracking(Base):
"""
AI 使用量追蹤表
追蹤 Gemini API 費用和所有 AI 使用統計
"""
__tablename__ = 'ai_usage_tracking'
id = Column(Integer, primary_key=True)
# AI 提供者: 'ollama', 'gemini'
provider = Column(String(20), nullable=False, index=True)
# 模型名稱: 'gemma3:4b', 'gemini-1.5-flash', 'gemini-2.5-pro'
model_name = Column(String(100), nullable=False)
# 使用類型: 'copy', 'web_search', 'product_insights', 'trend_keywords'
usage_type = Column(String(50), nullable=False)
# Token 用量
input_tokens = Column(Integer, default=0)
output_tokens = Column(Integer, default=0)
total_tokens = Column(Integer, default=0)
# 費用計算 (USD) - 主要用於 Gemini
input_cost = Column(Float, default=0.0)
output_cost = Column(Float, default=0.0)
total_cost = Column(Float, default=0.0)
# 響應時間 (秒)
duration = Column(Float)
# 請求資訊
request_date = Column(Date, nullable=False, index=True)
created_at = Column(DateTime, default=datetime.now)
created_by = Column(Integer, ForeignKey('users.id'))
# 關聯到歷史記錄 (可選)
history_id = Column(Integer, ForeignKey('ai_generation_history.id'))
__table_args__ = (
Index('idx_usage_provider_date', 'provider', 'request_date'),
Index('idx_usage_model_date', 'model_name', 'request_date'),
)
def to_dict(self):
"""轉換為字典格式"""
return {
'id': self.id,
'provider': self.provider,
'model_name': self.model_name,
'usage_type': self.usage_type,
'input_tokens': self.input_tokens,
'output_tokens': self.output_tokens,
'total_tokens': self.total_tokens,
'input_cost': self.input_cost,
'output_cost': self.output_cost,
'total_cost': self.total_cost,
'duration': self.duration,
'request_date': self.request_date.isoformat() if self.request_date else None,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
class AIPromptTemplate(Base):
"""AI 提示模板表 - 儲存常用的提示詞模板"""
__tablename__ = 'ai_prompt_templates'
id = Column(Integer, primary_key=True)
name = Column(String(100), nullable=False, unique=True) # 模板名稱
description = Column(String(255)) # 模板描述
template_type = Column(String(50), nullable=False, index=True) # copy, recommend, analysis
system_prompt = Column(Text) # 系統提示詞
user_prompt_template = Column(Text, nullable=False) # 用戶提示詞模板
# 預設參數
default_temperature = Column(Float, default=0.7)
default_style = Column(String(50))
is_active = Column(Boolean, default=True)
is_system = Column(Boolean, default=False) # 是否為系統內建
created_by = Column(Integer, ForeignKey('users.id'))
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
def to_dict(self):
"""轉換為字典格式"""
return {
'id': self.id,
'name': self.name,
'description': self.description,
'template_type': self.template_type,
'system_prompt': self.system_prompt,
'user_prompt_template': self.user_prompt_template,
'default_temperature': self.default_temperature,
'default_style': self.default_style,
'is_active': self.is_active,
'is_system': self.is_system,
'created_at': self.created_at.isoformat() if self.created_at else None,
}
# 預設的提示模板
DEFAULT_PROMPT_TEMPLATES = [
{
'name': '吸睛電商文案',
'description': '適合吸引眼球的促銷文案',
'template_type': 'copy',
'system_prompt': '''你是一位專業的電商銷售文案寫手,專門為台灣電商平台撰寫商品文案。
你的文案特點:
- 使用繁體中文
- 簡潔有力,通常在 100 字以內
- 善用表情符號增加吸引力
- 強調商品賣點和消費者利益
- 適時使用行動呼籲 (CTA)''',
'user_prompt_template': '''請為以下商品撰寫銷售文案:
商品名稱:{product_name}
文案風格:使用吸引眼球的標題和表情符號
{trend_context}
請生成一段吸引人的銷售文案100字以內''',
'default_temperature': 0.8,
'default_style': '吸睛',
'is_system': True,
},
{
'name': '專業產品介紹',
'description': '適合強調功效和成分的專業文案',
'template_type': 'copy',
'system_prompt': '''你是一位專業的產品行銷專家,擅長撰寫專業且有說服力的產品介紹。
你的文案特點:
- 使用繁體中文
- 強調產品的專業性和科學依據
- 使用精確的數據和專業術語
- 建立品牌信任感''',
'user_prompt_template': '''請為以下商品撰寫專業介紹:
商品名稱:{product_name}
文案風格:使用專業術語,強調成分和功效
{trend_context}
請生成一段專業的產品介紹100字以內''',
'default_temperature': 0.5,
'default_style': '專業',
'is_system': True,
},
{
'name': '限時促銷文案',
'description': '創造緊迫感的促銷文案',
'template_type': 'copy',
'system_prompt': '''你是一位擅長製造緊迫感的行銷文案專家。
你的文案特點:
- 使用繁體中文
- 善用限時、限量等字眼
- 創造錯過可惜的感覺
- 強調立即行動的好處''',
'user_prompt_template': '''請為以下商品撰寫限時促銷文案:
商品名稱:{product_name}
文案風格:使用限時優惠的語氣,創造緊迫感
{trend_context}
請生成一段有緊迫感的促銷文案100字以內''',
'default_temperature': 0.7,
'default_style': '急迫',
'is_system': True,
},
]
class AIInsight(Base):
"""
AI 洞察與知識庫表 (符合 ADR-007 雙寫規範)
Step 2 加入,供 OpenClaw 保存歷史 PPT、分析等輸出。
(embedding 欄位將在 Step 3 透過 SQL ALTER 增加,不宣告於 SQLAlchemy避免 SQLite 相容性錯誤)
"""
__tablename__ = 'ai_insights'
id = Column(Integer, primary_key=True, autoincrement=True)
insight_type = Column(String(50), nullable=False, index=True) # ppt, competitor_analysis, weekly_meta
period = Column(String(50), index=True) # 2026-04-16, 2026-W15
product_sku = Column(String(50), index=True) # 如果針對單一商品
content = Column(Text, nullable=False) # 具體輸出內容
metadata_json = Column(Text) # 附加元數據 (JSON 字串)
created_at = Column(DateTime, default=datetime.now, index=True)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
def to_dict(self):
import json
return {
'id': self.id,
'insight_type': self.insight_type,
'period': self.period,
'product_sku': self.product_sku,
'content': self.content,
'metadata': json.loads(self.metadata_json) if self.metadata_json else {},
'created_at': self.created_at.isoformat() if self.created_at else None,
}
def init_ai_tables(session):
"""
初始化 AI 相關表和預設資料
Args:
session: SQLAlchemy session
Returns:
tuple: (success: bool, message: str)
"""
try:
# 檢查是否已有預設模板
existing_count = session.query(AIPromptTemplate).filter_by(is_system=True).count()
if existing_count == 0:
# 新增預設模板
for template_data in DEFAULT_PROMPT_TEMPLATES:
template = AIPromptTemplate(**template_data)
session.add(template)
session.commit()
return True, f"AI 模板初始化完成,新增 {len(DEFAULT_PROMPT_TEMPLATES)} 個預設模板"
else:
return True, f"AI 模板已存在 ({existing_count} 個系統模板)"
except Exception as e:
session.rollback()
return False, f"AI 模板初始化失敗: {e}"