Files
qihuaiflow/backend/python/app/api/market.py
T

819 lines
28 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.
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from sqlalchemy import and_, desc, asc, func
from pydantic import BaseModel, Field
from typing import Optional
import uuid
from datetime import datetime
import httpx
from urllib.parse import urlparse
import re
from ..database import get_db
from ..models import (
User, MarketApp, AppPurchase, AppReview, AppInstallation, Order,
AdBanner, MCPServer, gen_uuid
)
from ..auth import get_current_user, get_current_user_id, get_current_user_id_optional
from ..middleware.log_service import LogService
from app.api.admin.mcp_admin import _test_mcp_server
router = APIRouter()
def _extract_url_from_text(text: Optional[str]) -> Optional[str]:
"""Extract first http/https URL from text."""
if not text:
return None
match = re.search(r"https?://[^\s\"'<>\)\]\}]+", text)
return match.group(0) if match else None
# ===== Agent 接入前连通性 / 鉴权校验 =====
def _normalize_dify_url(endpoint: str) -> str:
ep = endpoint.rstrip("/")
if ep.endswith("/chat-messages"):
return ep
if ep.endswith("/v1/chat-messages"):
return ep
if ep.endswith("/v1"):
return ep + "/chat-messages"
return ep + "/v1/chat-messages"
def _short_err(e: Exception) -> str:
return str(e)[:160]
def _post_follow(url: str, headers: dict, json_body: dict, timeout: float = 15.0):
"""POST 并手动跟随 301/302/307/308 重定向(**保持 POST 方法与请求体**)。
背景:Dify 若填写 http:// 地址会 301 永久重定向到 https://;httpx 内建
follow_redirects 对 301/302 会把 POST 降级为 GET(浏览器行为),导致对
/v1/chat-messages 变成 GET → 405 method_not_allowed。故手动跟随并保持 POST。
容器内 env 有坏代理(502),必须 trust_env=False 直连。
"""
cur = url
last = None
for _ in range(4):
last = httpx.post(cur, headers=headers, json=json_body, timeout=timeout,
trust_env=False, follow_redirects=False)
if last.status_code in (301, 302, 307, 308):
loc = last.headers.get("location")
if not loc:
return last
cur = loc if loc.startswith("http") else str(httpx.URL(cur).join(loc))
continue
return last
return last
def verify_agent_connectivity(platform: str, endpoint: str, api_key: str) -> dict:
"""
接入 Agent 前的连通性 + 鉴权校验。
返回 {ok, verified, detail}
ok: 是否通过(dify=鉴权通过;n8n/coze/其他=地址可达)
verified: 是否做了真正的鉴权验证(dify 为 True,其余为 False)
"""
ep = (endpoint or "").strip()
if not ep or not ep.startswith(("http://", "https://")):
return {"ok": False, "verified": False, "detail": "Endpoint 必须以 http:// 或 https:// 开头"}
platform = (platform or "dify").lower()
# 关键:容器内 env 有坏代理(502),必须 trust_env=False 直连
if platform == "dify":
url = _normalize_dify_url(ep)
payload = {
"inputs": {},
"query": "连通性测试",
"user": "__verify__",
"response_mode": "blocking",
}
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
try:
r = _post_follow(url, headers, payload, timeout=15.0)
except Exception as e:
return {"ok": False, "verified": False, "detail": f"无法连接 Dify 地址:{_short_err(e)}"}
if r.status_code == 200:
return {"ok": True, "verified": True, "detail": "连通且 API Key 有效"}
if r.status_code in (401, 403):
return {"ok": False, "verified": False, "detail": "API Key 无效(401/403)"}
if r.status_code == 404:
return {"ok": False, "verified": False,
"detail": "地址路径不正确(404),应为 .../v1/chat-messages"}
if r.status_code in (301, 302, 307, 308):
loc = r.headers.get("location") or ""
return {"ok": False, "verified": False,
"detail": f"地址被重定向({r.status_code})到 {loc or '未知'};"
f"请直接填写重定向后的地址(通常把 http:// 改为 https://)"}
if r.status_code == 400:
# 400 需区分两种情况:
# (a) 到达应用且鉴权通过,但 Dify 应用/工作流内部执行报错(invalid_param + Run failed)
# (b) 纯参数错误
body = (r.text or "")
low = body.lower()
if ("run failed" in low or "error code" in low or "workflow" in low
or "node" in low or "-500" in body):
return {"ok": True, "verified": True,
"detail": "连通且 API Key 有效;但你的 Dify 应用内部执行报错:"
f"{body[:180]}。这是你自建 Dify 应用/工作流本身的问题"
"(例如代码节点沙箱缺少 python3),请在 Dify 侧修复。"}
return {"ok": False, "verified": False, "detail": f"参数错误(400):{body[:180]}"}
return {"ok": False, "verified": False, "detail": f"Dify 返回状态码 {r.status_code}"}
# n8n / coze / 其他:仅做可达性探测(HEAD,避免触发 webhook 副作用)
try:
r = httpx.head(ep, timeout=12.0, trust_env=False, follow_redirects=True)
if r.status_code < 500:
return {"ok": True, "verified": False,
"detail": f"地址可连通(HTTP {r.status_code}),运行时再做协议级鉴权"}
return {"ok": False, "verified": False, "detail": f"服务返回 {r.status_code},地址可能不可用"}
except Exception as e:
return {"ok": False, "verified": False, "detail": f"地址不可达:{_short_err(e)}"}
class VerifyAgentRequest(BaseModel):
platform: Optional[str] = None
endpoint: str
api_key: str
@router.post("/apps/verify-agent")
def verify_agent(
req: VerifyAgentRequest,
current_user: User = Depends(get_current_user),
):
"""接入 Agent 前手动测试连通性 / 鉴权,供前台「测试连接」按钮调用。"""
result = verify_agent_connectivity(req.platform or "dify", req.endpoint, req.api_key)
return {"code": 0, "data": result, "message": "ok"}
@router.get("/ads")
def list_public_ads(db: Session = Depends(get_db)):
"""
前台首页「精选推荐」广告位:返回已上架(active)广告,按排序升序,无需登录。
供 LandingPage 轮播调用,无鉴权依赖。
"""
ads = (
db.query(AdBanner)
.filter(AdBanner.status == "active")
.order_by(AdBanner.sort_order.asc())
.all()
)
return [
{
"id": a.id,
"title": a.title,
"image_url": a.image_url,
"link_app_id": a.link_app_id,
"link_app_name": a.link_app_name,
"sort_order": a.sort_order,
"click_count": a.click_count,
"status": a.status,
"created_at": a.created_at.isoformat() if a.created_at else None,
}
for a in ads
]
class UploadAppRequest(BaseModel):
name: str = Field(min_length=1, max_length=200)
description: Optional[str] = None
icon: Optional[str] = None
type: Optional[str] = None # skill, agent, mcp
source: Optional[str] = None
price: float = Field(default=0, ge=0)
download_url: Optional[str] = None
category: Optional[str] = None
tags: Optional[str] = None
is_member_free: Optional[bool] = False
join_promotion: Optional[bool] = False
agent_endpoint: Optional[str] = None
api_key: Optional[str] = None
platform: Optional[str] = None
class PurchaseAppRequest(BaseModel):
amount: Optional[float] = None
class ReviewAppRequest(BaseModel):
rating: int = Field(ge=1, le=5)
content: Optional[str] = None
@router.get("/apps")
def list_apps(
type: Optional[str] = Query(None),
source: Optional[str] = Query(None),
category: Optional[str] = Query(None),
search: Optional[str] = Query(None),
sort: str = Query(default="latest"),
page: int = Query(default=1, ge=1),
page_size: int = Query(default=20, ge=1, le=100),
db: Session = Depends(get_db),
user_id: Optional[str] = Depends(get_current_user_id_optional),
):
query = db.query(MarketApp).filter(MarketApp.status == "published")
if type:
query = query.filter(MarketApp.type == type)
if source:
query = query.filter(MarketApp.source == source)
if category:
query = query.filter(MarketApp.type == category)
if search:
query = query.filter(
and_(
MarketApp.name.ilike(f"%{search}%")
| MarketApp.description.ilike(f"%{search}%")
)
)
if sort == "popular":
query = query.order_by(desc(MarketApp.install_count))
elif sort == "rating":
query = query.order_by(desc(MarketApp.rating_avg))
else:
query = query.order_by(desc(MarketApp.created_at))
total = query.count()
apps = query.offset((page - 1) * page_size).limit(page_size).all()
# 如果用户已登录,获取该用户已安装的应用ID集合(确保按用户隔离)
installed_app_ids = set()
if user_id:
installations = db.query(AppInstallation.app_id).filter(
AppInstallation.user_id == user_id
).all()
installed_app_ids = {inst[0] for inst in installations}
# 查询已安装状态时始终带 user_id 过滤
result = []
for app in apps:
# 用户自己创建/导入的应用应始终视为已安装(避免安装记录缺失导致显示未安装)
is_user_owned = bool(user_id) and app.source == "user" and app.publisher_id == user_id
item = {
"id": app.id,
"name": app.name,
"description": app.description,
"icon": app.icon,
"type": app.type,
"source": app.source,
"publisher_id": app.publisher_id,
"publisher_name": app.publisher_name,
"is_member_free": app.is_member_free if app.is_member_free else False,
"join_promotion": app.join_promotion if app.join_promotion else False,
"status": app.status,
"price": float(app.price) if app.price else 0,
"install_count": app.install_count or 0,
"rating_avg": float(app.rating_avg) if app.rating_avg else 0,
"review_count": app.review_count or 0,
"created_at": str(app.created_at) if app.created_at else None,
"installed": (app.id in installed_app_ids or is_user_owned) if user_id else False,
}
result.append(item)
return {
"code": 0,
"data": {
"total": total,
"page": page,
"page_size": page_size,
"items": result,
},
"message": "ok",
}
@router.get("/apps/{app_id}")
def get_app(
app_id: str,
db: Session = Depends(get_db),
current_user: Optional[User] = Depends(get_current_user),
):
app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
if not app:
raise HTTPException(status_code=404, detail="应用不存在")
# 检查当前用户是否已购买(按用户隔离)
purchased = False
installed = False
if current_user:
purchase = db.query(AppPurchase).filter(
and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
).first()
if purchase:
purchased = True
installed = purchase.status == "installed"
# 通过 AppInstallation 表检查是否已安装(按用户隔离)
installation = db.query(AppInstallation).filter(
and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id)
).first()
if installation:
installed = True
return {
"code": 0,
"data": {
"id": app.id, "name": app.name, "description": app.description,
"icon": app.icon, "type": app.type, "source": app.source,
"type": app.type,
"price": float(app.price) if app.price else 0,
"install_count": app.install_count or 0,
"rating_avg": float(app.rating_avg) if app.rating_avg else 0,
"review_count": app.review_count or 0,
"publisher_id": app.publisher_id,
"download_url": app.download_url,
"detail": app.detail,
"created_at": str(app.created_at),
"purchased": purchased,
"installed": installed,
},
"message": "ok",
}
@router.post("/apps")
def upload_app(
req: UploadAppRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
# 接入 Agent 时先做连通性 / 鉴权校验,避免把无效连接存进库
if (req.type or "") == "agent" and req.agent_endpoint and req.api_key:
v = verify_agent_connectivity(req.platform or "dify", req.agent_endpoint, req.api_key)
if not v["ok"]:
raise HTTPException(
status_code=400,
detail=f"Agent 连接校验未通过:{v['detail']}",
)
# source 判定:管理员(admin 角色)通过前台市场导入/开发的 Agent/Skill/MCP
# 统一标记为官方(official),其余用户按请求或默认为 user。
if current_user.role == "admin":
app_source = "official"
elif req.source and req.source in ("official", "user"):
app_source = req.source
else:
app_source = "user"
app = MarketApp(
id=gen_uuid(),
name=req.name,
description=req.description,
icon=req.icon,
type=req.type if req.type in ("skill","agent","mcp") else "skill",
source=app_source,
platform=req.platform,
price=req.price,
install_count=0,
rating_avg=0,
review_count=0,
publisher_id=current_user.id,
publisher_name=current_user.username,
is_member_free=req.is_member_free if req.is_member_free is not None else False,
join_promotion=req.join_promotion if req.join_promotion is not None else False,
agent_endpoint=req.agent_endpoint,
agent_api_key=req.api_key,
status="published", # 用户创建的应用直接发布
created_at=datetime.now(),
updated_at=datetime.now(),
)
db.add(app)
db.commit()
db.refresh(app)
# MCP 应用同步注册到后台 MCP Client 管理(mcp_servers 表)
if app.type == "mcp":
transport = "streamable_http"
code = req.download_url or ""
if code:
app.skill_content = code
app.mcp_transport = transport
db.commit()
db.refresh(app)
existing_server = db.query(MCPServer).filter(MCPServer.id == app.id).first()
if not existing_server:
url = _extract_url_from_text(code) or _extract_url_from_text(req.description or "") or ""
server = MCPServer(
id=app.id,
name=app.name,
description=app.description or "",
transport_type=transport,
command="",
args="{}",
env_vars="{}",
url=url,
headers="{}",
is_enabled=True,
status="disconnected",
tools_count=0,
resources_count=0,
created_at=app.created_at or datetime.now(),
updated_at=app.updated_at or datetime.now(),
)
db.add(server)
db.commit()
# 导入后自动做一次连通性测试,状态会同步为 connected / error
_test_mcp_server(server, db)
# 用户创建/导入的应用自动标记为已安装
existing_install = db.query(AppInstallation).filter(
AppInstallation.user_id == current_user.id,
AppInstallation.app_id == app.id,
).first()
if not existing_install:
installation = AppInstallation(
user_id=current_user.id,
app_id=app.id,
installed_at=datetime.now(),
)
db.add(installation)
db.commit()
LogService.log_operation(
user_id=current_user.id, username=current_user.username, action="upload_app",
target="market_app", target_id=app.id,
detail=f"上传应用 {app.name}",
)
return {
"code": 0,
"data": {
"id": app.id, "name": app.name, "status": app.status,
"created_at": str(app.created_at),
},
"message": "提交成功,等待审核",
}
@router.post("/apps/{app_id}/purchase")
def purchase_app(
app_id: str,
req: PurchaseAppRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
app = db.query(MarketApp).filter(
and_(MarketApp.id == app_id, MarketApp.status == "published")
).first()
if not app:
raise HTTPException(status_code=404, detail="应用不存在或未上架")
# 检查是否已购买
existing = db.query(AppPurchase).filter(
and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
).first()
if existing and existing.status in ("purchased", "installed"):
raise HTTPException(status_code=400, detail="您已购买该应用")
amount = req.amount or app.price
# 创建订单
order = Order(
id=gen_uuid(),
user_id=current_user.id,
order_type="app",
item_id=app_id,
item_name=app.name,
amount=amount,
status="pending",
created_at=datetime.now(),
)
db.add(order)
db.flush()
# 创建购买记录
purchase = AppPurchase(
id=gen_uuid(),
user_id=current_user.id,
app_id=app_id,
order_id=order.id,
status="purchased",
purchased_at=datetime.now(),
)
db.add(purchase)
# 更新应用下载量
app.install_count = (app.install_count or 0) + 1
app.updated_at = datetime.now()
db.commit()
return {
"code": 0,
"data": {
"order_id": order.id,
"purchase_id": purchase.id,
"amount": amount,
"status": "purchased",
},
"message": "购买成功",
}
@router.post("/apps/{app_id}/install")
def install_app(
app_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
if not app:
raise HTTPException(status_code=404, detail="应用不存在")
# 检查是否已安装(通过 AppInstallation 表)
existing_install = db.query(AppInstallation).filter(
and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id)
).first()
if existing_install:
raise HTTPException(status_code=400, detail="应用已安装")
# 检查是否已购买(付费应用需要先购买)
purchase = db.query(AppPurchase).filter(
and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
).first()
# 对于付费应用,必须已购买才能安装
if float(app.price or 0) > 0:
if not purchase:
raise HTTPException(status_code=403, detail="请先购买应用")
if purchase.status == "installed":
raise HTTPException(status_code=400, detail="应用已安装")
# 创建安装记录
installation = AppInstallation(
id=gen_uuid(),
user_id=current_user.id,
app_id=app_id,
installed_at=datetime.now(),
)
db.add(installation)
# 如果有购买记录,更新状态为已安装
if purchase:
purchase.status = "installed"
purchase.installed_at = datetime.now()
# 更新应用安装量
app.install_count = (app.install_count or 0) + 1
app.updated_at = datetime.now()
db.commit()
return {
"code": 0,
"data": {"installation_id": installation.id, "status": "installed"},
"message": "安装成功",
}
@router.delete("/apps/{app_id}/uninstall")
def uninstall_app(
app_id: str,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""卸载应用 - 删除当前用户的安装记录(按用户隔离)"""
app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
if not app:
raise HTTPException(status_code=404, detail="应用不存在")
# 查找当前用户的安装记录(按用户隔离)
installation = db.query(AppInstallation).filter(
and_(AppInstallation.app_id == app_id, AppInstallation.user_id == current_user.id)
).first()
if not installation:
raise HTTPException(status_code=400, detail="应用未安装")
# 删除安装记录
db.delete(installation)
# 问题4修复: 将AppPurchase状态从installed重置为purchased
purchase = db.query(AppPurchase).filter(
and_(AppPurchase.app_id == app_id, AppPurchase.user_id == current_user.id)
).first()
if purchase and purchase.status == "installed":
purchase.status = "purchased"
# 更新应用安装量(减1,最小为0)
if app.install_count and app.install_count > 0:
app.install_count = app.install_count - 1
app.updated_at = datetime.now()
db.commit()
LogService.log_operation(
user_id=current_user.id, username=current_user.username, action="uninstall_app",
target="market_app", target_id=app_id,
detail=f"卸载应用 {app.name}",
)
return {
"code": 0,
"message": "卸载成功",
}
@router.post("/apps/{app_id}/review")
def review_app(
app_id: str,
req: ReviewAppRequest,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
app = db.query(MarketApp).filter(MarketApp.id == app_id).first()
if not app:
raise HTTPException(status_code=404, detail="应用不存在")
# 检查是否已评价
existing = db.query(AppReview).filter(
and_(AppReview.app_id == app_id, AppReview.user_id == current_user.id)
).first()
if existing:
# 更新已有评价
old_rating = existing.rating
existing.rating = req.rating
existing.content = req.content
existing.updated_at = datetime.now()
# 重新计算平均分 (总数不变,减去旧评分加新评分)
if app.review_count and app.review_count > 0:
total_rating = float(app.rating_avg or 0) * app.review_count
total_rating = total_rating - old_rating + req.rating
app.rating_avg = total_rating / app.review_count
else:
app.rating_avg = req.rating
app.review_count = 1
else:
# 创建新评价
review = AppReview(
id=gen_uuid(),
app_id=app_id,
user_id=current_user.id,
rating=req.rating,
content=req.content,
created_at=datetime.now(),
updated_at=datetime.now(),
)
db.add(review)
# 重新计算平均分
app.review_count = (app.review_count or 0) + 1
if app.review_count == 1:
app.rating_avg = req.rating
else:
total_rating = float(app.rating_avg or 0) * (app.review_count - 1) + req.rating
app.rating_avg = round(total_rating / app.review_count, 1)
app.updated_at = datetime.now()
db.commit()
return {
"code": 0,
"data": {
"rating_avg": float(app.rating_avg or 0),
"review_count": app.review_count or 0,
},
"message": "评价成功",
}
@router.get("/my-apps")
def get_my_apps(
status: Optional[str] = Query(None),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
query = db.query(AppPurchase).filter(AppPurchase.user_id == current_user.id)
purchases = query.order_by(desc(AppPurchase.created_at)).all()
result = []
for p in purchases:
app = db.query(MarketApp).filter(MarketApp.id == p.app_id).first()
if app:
result.append({
"id": p.id,
"app_id": app.id,
"name": app.name,
"icon": app.icon,
"type": app.type,
"price": float(app.price or 0),
"purchased_at": str(p.created_at) if p.created_at else None,
})
return {"code": 0, "data": result, "message": "ok"}
@router.get("/reviews/{app_id}")
def get_reviews(
app_id: str,
page: int = Query(default=1, ge=1),
page_size: int = Query(default=20, ge=1, le=100),
db: Session = Depends(get_db),
):
query = db.query(AppReview).filter(AppReview.app_id == app_id)
total = query.count()
reviews = (
query.order_by(desc(AppReview.created_at))
.offset((page - 1) * page_size)
.limit(page_size)
.all()
)
result = []
for r in reviews:
user = db.query(User).filter(User.id == r.user_id).first()
result.append({
"id": r.id,
"user_id": r.user_id,
"username": user.username if user else "unknown",
"avatar": user.avatar if user else None,
"rating": r.rating,
"content": r.content,
"created_at": str(r.created_at) if r.created_at else None,
})
return {
"code": 0,
"data": {"total": total, "page": page, "page_size": page_size, "items": result},
"message": "ok",
}
# ============================================================
# Agent部署资源查询接口
# ============================================================
@router.get("/agent-resources")
def get_agent_resources(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Agent部署时需要的资源列表:可用的应用、知识库、文档"""
# 用户已发布的应用
my_apps = db.query(MarketApp).filter(
MarketApp.publisher_id == current_user.id,
MarketApp.status == "published",
).order_by(desc(MarketApp.updated_at)).all()
apps_result = []
for app in my_apps:
apps_result.append({
"id": app.id,
"name": app.name,
"type": app.type,
"description": app.description,
"install_count": app.install_count or 0,
})
# 用户知识库
from ..models import KnowledgeBase, KnowledgeDocument
kb_list = db.query(KnowledgeBase).filter(
KnowledgeBase.user_id == current_user.id
).order_by(desc(KnowledgeBase.updated_at)).all()
kb_result = []
for kb in kb_list:
doc_count = db.query(KnowledgeDocument).filter(
KnowledgeDocument.knowledge_base_id == kb.id
).count()
kb_result.append({
"id": kb.id,
"name": kb.name,
"description": kb.description,
"document_count": doc_count,
})
# 用户已安装的应用(从AppPurchase)
installed = db.query(AppPurchase).filter(
AppPurchase.user_id == current_user.id
).all()
installed_ids = [p.app_id for p in installed]
installed_apps = []
if installed_ids:
iapps = db.query(MarketApp).filter(MarketApp.id.in_(installed_ids)).all()
for ia in iapps:
installed_apps.append({
"id": ia.id,
"name": ia.name,
"type": ia.type,
"description": ia.description,
})
return {
"code": 0,
"data": {
"my_apps": apps_result,
"knowledge_bases": kb_result,
"installed_apps": installed_apps,
},
"message": "ok",
}