import os, json, logging, httpx, sqlite3, uuid, base64, re, subprocess
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, JSONResponse
import uvicorn
from datetime import datetime, timezone, timedelta
app = FastAPI()
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(name)s] %(message)s")
logger = logging.getLogger("agent")
API_KEY = "sk-bbca4a0380d549389f0d27cdea0b5228"
API_URL = "https://api.deepseek.com/v1/chat/completions"
MODEL = "deepseek-v4-flash"
GITEA_TOKEN = "513f4f5151daf5754069f5c7dc0bfe6283989ef1"
GITEA_BASE = "https://gitea.xybkwd.top"
MEMOS_TOKEN = "memos_pat_GtKuYpUlRJP7KjWsCcBK51u87CS1f0CQ"
MEMOS_URL = "http://localhost:5230"
DB_PATH = "/home/ubuntu/ai-chat/chat.db"
BLOG_RSS = "https://www.xybkwd.top/rss.xml"
IC_DIR = "/home/ubuntu/inspiration-collector"
TZ_BJ = timezone(timedelta(hours=8))
SYSTEM_PROMPT = """你是冯总部署在东京服务器上的AI智能体。
## 冯总
- 33岁，省供电公司工会副主席
- 量化交易者，ETH/USDT永续合约
- freqtrade v2.2d策略运行中，Al Brooks价格行为学
## 你的能力
你可以通过以下命令执行操作（用户在对话中说相关的指令即可）：
### 读取类
- "查memos/最近的灵感" → 列出最近7天的灵感记录
- "读gitea/读文库/看笔记" → 读取Gitea仓库中的文件
- "看博客/读博客" → 读取博客最新文章
### 操作类
- "运行灵感分析/跑日报" → 立即运行daily_digest.py并推送到Gitea
### 服务器类
- "看状态/检查服务器" → 查看Docker容器和服务器状态
当用户提出上述需求时，直接执行并返回结果。
回答朴实有内容，北京时间UTC+8。"""
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("CREATE TABLE IF NOT EXISTS conversations (id TEXT PRIMARY KEY, title TEXT, created_at TEXT, updated_at TEXT)")
conn.execute("CREATE TABLE IF NOT EXISTS messages (id INTEGER PRIMARY KEY AUTOINCREMENT, conv_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT, created_at TEXT)")
conn.commit()
conn.close()
init_db()
# ============ TOOL FUNCTIONS ============
def tool_memos_list(days=7):
"""List recent memos."""
try:
import requests
payload = {"pageSize": 50}
r = requests.post(MEMOS_URL + "/memos.api.v1.MemoService/ListMemos",
json=payload, headers={"Authorization": "Bearer " + MEMOS_TOKEN, "Content-Type": "application/json"}, timeout=10)
memos = r.json().get("memos", [])
since = datetime.now(timezone.utc) - timedelta(days=days)
results = []
for m in memos:
ct = m.get("createTime", "")
try:
dt = datetime.fromisoformat(ct.replace("Z", "+00:00"))
except:
dt = datetime.now(timezone.utc)
if dt < since: continue
bj = dt + timedelta(hours=8)
results.append(f"[{bj.strftime('%m/%d %H:%M')}] {m.get('content','')}")
if not results:
return f"最近{days}天没有灵感记录。"
return "## 最近的灵感记录\n\n" + "\n\n".join(results[-20:])
except Exception as e:
return f"查询Memos失败: {e}"
def tool_gitea_list(repo, path=""):
"""List files in a Gitea repo."""
try:
url = f"{GITEA_BASE}/api/v1/repos/{repo}/contents/{path}"
r = httpx.get(url, headers={"Authorization": "token " + GITEA_TOKEN}, timeout=15)
data = r.json()
if isinstance(data, list):
items = [f"- [{i['type']}] {i['name']}" for i in data]
return f"## {repo}/{path}\n\n" + "\n".join(items)
elif isinstance(data, dict) and "message" in data:
return f"Gitea错误: {data['message']}"
return str(data)[:2000]
except Exception as e:
return f"Gitea读取失败: {e}"
def tool_gitea_read(repo, path, ref="main"):
"""Read a file from Gitea."""
try:
url = f"{GITEA_BASE}/api/v1/repos/{repo}/contents/{path.replace('/', '%2F')}?ref={ref}"
r = httpx.get(url, headers={"Authorization": "token " + GITEA_TOKEN}, timeout=15)
data = r.json()
if isinstance(data, dict) and "content" in data:
decoded = base64.b64decode(data["content"]).decode("utf-8")
if len(decoded) > 3000:
decoded = decoded[:3000] + f"\n\n... [截断，全文{len(decoded)}字]"
return decoded
elif isinstance(data, dict) and "message" in data:
return f"Gitea错误: {data['message']}"
return str(data)[:2000]
except Exception as e:
return f"Gitea读取失败: {e}"
def tool_blog_latest(count=5):
"""Read latest blog posts from RSS."""
try:
import feedparser
feed = feedparser.parse(BLOG_RSS)
entries = feed.entries[:count]
if not entries:
return "博客暂无文章。"
results = []
for e in entries:
results.append(f"### {e.title}\n- 时间: {e.get('published', '?')}\n- 链接: {e.link}\n- 摘要: {e.get('summary', '')[:300]}")
return "## 最新博客文章\n\n" + "\n\n".join(results)
except Exception as e:
return f"读取博客失败: {e}"
def tool_run_analysis():
"""Run the daily digest and push to Gitea."""
try:
script = os.path.join(IC_DIR, "analyzers", "daily_digest.py")
if not os.path.exists(script):
return f"分析脚本不存在: {script}"
result = subprocess.run(["python3", script], cwd=IC_DIR, capture_output=True, text=True, timeout=120)
output = (result.stdout or "") + (result.stderr or "")
# Extract key info
lines = [l for l in output.split("\n") if "written" in l.lower() or "done" in l.lower() or "pushed" in l.lower() or "error" in l.lower()]
summary = "\n".join(lines) if lines else output[:500]
return f"灵感分析已运行完成。\n{summary}\n\n查看: https://gitea.xybkwd.top/fxy/inspiration-collector/src/branch/main/ai-insights/daily/"
except subprocess.TimeoutExpired:
return "分析超时（>120秒），可能输出较长，请稍后检查Gitea。"
except Exception as e:
return f"运行分析失败: {e}"
def tool_server_status():
"""Check server status."""
try:
result = []
# docker
dp = subprocess.run("docker ps --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}'", shell=True, capture_output=True, text=True, timeout=10)
result.append("### Docker\n" + (dp.stdout or dp.stderr))
# disk/memory
df = subprocess.run("df -h / | tail -1", shell=True, capture_output=True, text=True, timeout=5)
result.append("### 磁盘\n" + (df.stdout or ""))
fm = subprocess.run("free -h | head -2", shell=True, capture_output=True, text=True, timeout=5)
result.append("### 内存\n" + (fm.stdout or ""))
return "\n".join(result)
except Exception as e:
return f"状态查询失败: {e}"
# ============ INTENT DETECTION ============
TOOL_PATTERNS = [
(r"(memos|灵感|最近记录|随手记)", lambda m: tool_memos_list(7)),
(r"(读一读|看一.*gitea|gitea.*文件|obsidian.*笔记|文库)", lambda m: tool_gitea_list("fxy/obsidian-vault")),
(r"(gitea.*obsidian|obsidian.*目录)", lambda m: tool_gitea_list("fxy/obsidian-vault")),
(r"(gitea.*灵感|灵感.*目录|分析.*目录)", lambda m: tool_gitea_list("fxy/inspiration-collector/ai-insights/daily")),
(r"(博客|blog|xybkwd|博文)", lambda m: tool_blog_latest(5)),
(r"(运行分析|跑日报|执行分析|run.*daily|run.*digest)", lambda m: tool_run_analysis()),
(r"(服务器状态|检查状态|docker|容器|磁盘)", lambda m: tool_server_status()),
]
def detect_intent(msg):
"""Detect if message matches a tool pattern. Returns result or None."""
msg_lower = msg.lower()
for pattern, handler in TOOL_PATTERNS:
if re.search(pattern, msg_lower):
logger.info("Intent matched: %s", pattern)
return handler(msg)
return None
# ============ HTML PAGE ============
INDEX_HTML = """
AI 智能体
"""
@app.get("/")
async def root():
h = {"Cache-Control": "no-cache, no-store, must-revalidate"}
return HTMLResponse(INDEX_HTML, headers=h)
@app.get("/convs")
async def list_convs():
conn = sqlite3.connect(DB_PATH)
rows = conn.execute("SELECT id, title FROM conversations ORDER BY updated_at DESC LIMIT 50").fetchall()
conn.close()
return {"convs": [{"id": r[0], "title": r[1]} for r in rows]}
@app.get("/convs/{conv_id}/messages")
async def get_messages(conv_id: str):
conn = sqlite3.connect(DB_PATH)
rows = conn.execute("SELECT role, content FROM messages WHERE conv_id=? ORDER BY id", (conv_id,)).fetchall()
conn.close()
return {"messages": [{"role": r[0], "content": r[1]} for r in rows]}
@app.delete("/convs/{conv_id}")
async def delete_conv(conv_id: str):
conn = sqlite3.connect(DB_PATH)
conn.execute("DELETE FROM messages WHERE conv_id=?", (conv_id,))
conn.execute("DELETE FROM conversations WHERE id=?", (conv_id,))
conn.commit()
conn.close()
return {"deleted": conv_id}
@app.post("/chat")
async def chat(req: Request):
body = await req.json()
msg = body.get("message", "")
conv_id = body.get("conv_id")
logger.info("User[%s]: %s", conv_id or "new", msg[:80])
# Save message to DB
conn = sqlite3.connect(DB_PATH)
if not conv_id:
conv_id = str(uuid.uuid4())[:8]
now = datetime.now(TZ_BJ).strftime("%Y-%m-%d %H:%M:%S")
title = (msg.strip()[:30] + "...") if len(msg.strip()) > 30 else msg.strip()
conn.execute("INSERT INTO conversations (id,title,created_at,updated_at) VALUES (?,?,?,?)", (conv_id, title, now, now))
conn.execute("INSERT INTO messages (conv_id,role,content) VALUES (?,'user',?)", (conv_id, msg))
conn.execute("UPDATE conversations SET updated_at=datetime('now','localtime') WHERE id=?", (conv_id,))
conn.commit()
conn.close()
# STEP 1: Try intent detection for tool commands
tool_result = detect_intent(msg)
tool_name = None
if tool_result:
# Determine which tool was used
for pattern, handler in TOOL_PATTERNS:
if re.search(pattern, msg.lower()) and handler(msg) == tool_result:
tool_name = pattern
break
# Simplify tool name for display
if "memos" in msg.lower() or "灵感" in msg:
tool_name = "查Memos"
elif "gitea" in msg.lower() or "文库" in msg or "笔记" in msg or "obsidian" in msg.lower():
tool_name = "读Gitea"
elif "博客" in msg or "blog" in msg.lower():
tool_name = "看博客"
elif "分析" in msg or "日报" in msg or "run" in msg.lower():
tool_name = "运行分析"
elif "状态" in msg or "docker" in msg.lower() or "磁盘" in msg:
tool_name = "服务器状态"
else:
tool_name = "工具"
# Augment tool result with AI commentary
try:
payload = {
"model": MODEL,
"messages": [
{"role": "system", "content": "你是一个智能体。用户执行了一个工具操作，以下是工具返回的结果。请用自然语言总结回复用户，指出关键信息。回复要简洁有实质。"},
{"role": "user", "content": f"工具({tool_name})返回结果:\n\n{tool_result}"}
],
"temperature": 0.3, "max_tokens": 1024
}
headers = {"Authorization": "Bearer " + API_KEY, "Content-Type": "application/json"}
async with httpx.AsyncClient(timeout=60) as client:
r = await client.post(API_URL, json=payload, headers=headers)
data = r.json()
reply = data.get("choices", [{}])[0].get("message", {}).get("content", "")
if not reply:
reply = tool_result[:2000]
except Exception:
reply = tool_result[:2000]
# Save reply
conn2 = sqlite3.connect(DB_PATH)
conn2.execute("INSERT INTO messages (conv_id,role,content) VALUES (?,'assistant',?)", (conv_id, reply))
conn2.execute("UPDATE conversations SET updated_at=datetime('now','localtime') WHERE id=?", (conv_id,))
conn2.commit()
conn2.close()
logger.info("Tool[%s]: %s", tool_name, msg[:60])
return {"reply": reply, "conv_id": conv_id, "tool": tool_name}
# STEP 2: No tool intent — normal chat
conn3 = sqlite3.connect(DB_PATH)
rows = conn3.execute("SELECT role,content FROM messages WHERE conv_id=? ORDER BY id DESC LIMIT 10", (conv_id,)).fetchall()
rows.reverse()
conn3.close()
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
for r in rows:
messages.append({"role": r[0], "content": r[1]})
try:
payload = {"model": MODEL, "messages": messages, "temperature": 0.6, "max_tokens": 4096}
headers = {"Authorization": "Bearer " + API_KEY, "Content-Type": "application/json"}
async with httpx.AsyncClient(timeout=120) as client:
r = await client.post(API_URL, json=payload, headers=headers)
data = r.json()
reply = data.get("choices", [{}])[0].get("message", {}).get("content", "")
conn4 = sqlite3.connect(DB_PATH)
conn4.execute("INSERT INTO messages (conv_id,role,content) VALUES (?,'assistant',?)", (conv_id, reply))
conn4.execute("UPDATE conversations SET updated_at=datetime('now','localtime') WHERE id=?", (conv_id,))
conn4.commit()
conn4.close()
logger.info("Chat[%s]: %d chars", conv_id, len(reply))
return {"reply": reply, "conv_id": conv_id}
except Exception as e:
logger.error("Error: %s", e)
return JSONResponse({"error": str(e)}, status_code=500)
if __name__ == "__main__":
uvicorn.run(app, host="127.0.0.1", port=8900)