Upload 13 files

app.py

@@ -1,22 +1,31 @@
 # ⚙️ 后端逻辑/核心服务端.py (Hugging Face Spaces app.py)
-from fastapi import FastAPI, File, UploadFile, Form
+from fastapi import FastAPI, File, UploadFile, Form, Depends
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import Response, JSONResponse
-from pydantic import BaseModel  # 【新增】：引入标准数据模型
+from sqlalchemy.orm import Session
+from pydantic import BaseModel
+from huggingface_hub import hf_hub_download, HfApi
 import hashlib
 import urllib.parse
 import urllib.request
 import os
 import 数据库连接 as db
 
-# 引入拆分后的四大业务模块
 from router_users import router as users_router
 from router_items import router as items_router
 from router_comments import router as comments_router
 from router_messages import router as messages_router
+from router_wallet import router as wallet_router
+from database_sql import init_sql_db, get_db
+from models_sql import Ownership
 
 app = FastAPI(title="ComfyUI Ranking Community API")
 
+@app.on_event("startup")
+def on_startup():
+    init_sql_db()
+    print("关系型数据库加载完毕，金融表同步完成。")
+
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -29,10 +38,11 @@ app.include_router(users_router)
 app.include_router(items_router)
 app.include_router(comments_router)
 app.include_router(messages_router)
+app.include_router(wallet_router)
 
 @app.get("/")
 def read_root(): 
-    return {"status": "ok", "message": "API is running and fully modularized!"}
+    return {"status": "ok"}
 
 @app.post("/api/upload")
 async def upload_file(file: UploadFile = File(...), file_type: str = Form(...)):
@@ -52,36 +62,82 @@ async def upload_file(file: UploadFile = File(...), file_type: str = Form(...)):
     url = f"https://huggingface.co/datasets/{db.DATASET_REPO_ID}/resolve/main/{target_dir}/{safe_url_filename}"
     return {"status": "success", "url": url, "display_name": file.filename, "hashed_name": new_filename}
 
+# =======================================================
+# 【核心新增】：死链与资源有效性检测 (解决问题1：防买空)
+# =======================================================
+class ValidateRequest(BaseModel):
+    item_id: str
 
-# =========================================================
-# 【核心修复】：使用 huggingface_hub 官方库替代 urllib 解决重定向 401 问题
-# =========================================================
-from huggingface_hub import hf_hub_download
+@app.post("/api/validate_resource")
+async def validate_resource(req: ValidateRequest):
+    items_db = db.load_data("items.json", default_data=[])
+    item = next((i for i in items_db if i["id"] == req.item_id), None)
+    if not item:
+        return JSONResponse(content={"error": "该资源已被原作者删除"}, status_code=404)
+        
+    link = item.get("link", "")
+    itype = item.get("type", "")
+    
+    if itype.startswith("tool"):
+        # 探测 Git 仓库是否为 404 死链
+        try:
+            req_obj = urllib.request.Request(link, method="HEAD", headers={'User-Agent': 'Mozilla/5.0'})
+            with urllib.request.urlopen(req_obj, timeout=5) as response:
+                if response.status >= 400:
+                    return JSONResponse(content={"error": "原作者的 Git 仓库已失效或设为私有"}, status_code=400)
+        except Exception as e:
+            return JSONResponse(content={"error": "原作者的 Git 仓库无法访问，链接已失效"}, status_code=400)
+            
+    elif itype.startswith("app"):
+        # 探测 HF 云端的 JSON 文件是否丢失
+        if "resolve/main/" in link:
+            repo_path = urllib.parse.unquote(link.split("resolve/main/")[-1])
+            hf_token = os.environ.get("HF_TOKEN")
+            try:
+                api = HfApi()
+                exists = api.file_exists(repo_id=db.DATASET_REPO_ID, filename=repo_path, repo_type="dataset", token=hf_token)
+                if not exists:
+                    return JSONResponse(content={"error": "该工作流的 JSON 文件已在云端损坏或丢失"}, status_code=400)
+            except Exception:
+                pass # 忽略 HF API 自身的偶发网络波动，不强行阻断
+                
+    return {"status": "success"}
 
+# =======================================================
+# 代理下载与所有权鉴权防线
+# =======================================================
 class ProxyDownloadRequest(BaseModel):
     url: str
+    item_id: str
+    account: str
 
 @app.post("/api/proxy_download")
-async def proxy_download(req_data: ProxyDownloadRequest):
+async def proxy_download(req_data: ProxyDownloadRequest, sql_db: Session = Depends(get_db)):
     target_url = req_data.url
-    
-    # 校验 URL 格式
     if not target_url or "resolve/main/" not in target_url:
         return JSONResponse(content={"error": "无效的 Hugging Face 下载链接"}, status_code=400)
         
+    items_db = db.load_data("items.json", default_data=[])
+    item = next((i for i in items_db if i["id"] == req_data.item_id), None)
+    
+    if not item: return JSONResponse(content={"error": "资源不存在或已被删除"}, status_code=404)
+        
+    price = int(item.get("price", 0))
+    author = item.get("author")
+    
+    # 所有权拦截：如果收费且不是作者本人，严查 SQL 所有权表
+    if price > 0 and req_data.account != author:
+        owned = sql_db.query(Ownership).filter(Ownership.account == req_data.account, Ownership.item_id == req_data.item_id).first()
+        if not owned:
+            return JSONResponse(content={"error": "🚨 非法下载：云端数据库未找到您的购买凭证！"}, status_code=403)
+
     hf_token = os.environ.get("HF_TOKEN")
-    if not hf_token:
-        return JSONResponse(content={"error": "云端环境变量未配置 HF_TOKEN，无法读取私有库"}, status_code=401)
+    if not hf_token: return JSONResponse(content={"error": "云端环境变量未配置 HF_TOKEN"}, status_code=401)
         
     try:
-        # 1. 从前端传来的 URL 中切割出仓库内的真实相对路径
-        # 例如从 https://.../resolve/main/apps/123_%E6%B5%8B.json 提取出 apps/123_%E6%B5%8B.json
         repo_path_encoded = target_url.split("resolve/main/")[-1]
-        
-        # 2. 对可能存在的被编码的中文字符进行解码 (恢复为 apps/123_测.json)
         repo_path = urllib.parse.unquote(repo_path_encoded)
         
-        # 3. 调用官方库直接从私有 Dataset 拉取文件（完美处理鉴权和重定向）
         cached_file_path = hf_hub_download(
             repo_id=db.DATASET_REPO_ID,
             repo_type="dataset",
@@ -89,7 +145,6 @@ async def proxy_download(req_data: ProxyDownloadRequest):
             token=hf_token
         )
         
-        # 4. 读取缓存在云端的本地文件流并返回给前端
         with open(cached_file_path, "rb") as f:
             content = f.read()
             

database_sql.py

@@ -0,0 +1,26 @@
+# database_sql.py
+import os
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from models_sql import Base
+
+# 优先读取环境变量中的 PostgreSQL 数据库连接，如果没有则使用 SQLite 降级方案
+# 生产环境建议在 HF Spaces 的 Secrets 中配置 DATABASE_URL = postgresql://user:pass@host/dbname
+SQLALCHEMY_DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:////tmp/comfy_financial.db")
+
+# 如果是 SQLite，需要 check_same_thread=False
+connect_args = {"check_same_thread": False} if "sqlite" in SQLALCHEMY_DATABASE_URL else {}
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args=connect_args)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+def init_sql_db():
+    # 启动时自动建表
+    Base.metadata.create_all(bind=engine)
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

models.py

@@ -63,7 +63,7 @@ class ItemCreate(BaseModel):
     link: str
     coverUrl: Optional[str] = None
     author: str 
-    price: float = 0.0
+    price: int = 0
 
 class FollowToggle(BaseModel):
     user_id: str
@@ -87,4 +87,20 @@ class ItemUpdate(BaseModel):
     fullDesc: Optional[str] = None
     link: Optional[str] = None
     coverUrl: Optional[str] = None
-    price: Optional[float] = None
+    price: Optional[int] = None
+
+class RechargeRequest(BaseModel):
+    account: str
+    amount: int       # 充值的积分数量
+    method: str       # "alipay" 或 "wechat"
+
+class WithdrawRequest(BaseModel):
+    account: str
+    amount: int       # 提现积分数量
+    alipay_account: str
+    real_name: str
+    code: str         # 邮箱安全验证码
+
+class PurchaseRequest(BaseModel):
+    account: str
+    item_id: str

models_sql.py

@@ -0,0 +1,43 @@
+# models_sql.py
+from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Float
+from sqlalchemy.orm import declarative_base
+import datetime
+
+Base = declarative_base()
+
+class Wallet(Base):
+    """用户钱包表"""
+    __tablename__ = "wallets"
+    
+    account = Column(String, primary_key=True, index=True)
+    balance = Column(Integer, default=0)         # 用于消费的余额 (充值获得)
+    earn_balance = Column(Integer, default=0)    # 创作者收益余额 (别人购买获得)
+    frozen_balance = Column(Integer, default=0)  # 提现审核冻结中的余额
+    
+    # 乐观锁版本号，防止并发扣款被击穿
+    version = Column(Integer, default=1)
+
+class Ownership(Base):
+    """资源所有权表 (解决Req 4：一次购买，终身免费)"""
+    __tablename__ = "ownerships"
+    
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    account = Column(String, index=True)
+    item_id = Column(String, index=True)
+    purchased_at = Column(DateTime, default=datetime.datetime.utcnow)
+
+class Transaction(Base):
+    """交易流水表 (复式记账，解决Req 6, 7：税务对账与哈希防篡改)"""
+    __tablename__ = "transactions"
+    
+    tx_id = Column(String, primary_key=True)  # 全局唯一流水号 (如 ORDER_uuid)
+    account = Column(String, index=True)
+    tx_type = Column(String)                  # RECHARGE, CONSUME, EARN, WITHDRAW
+    amount = Column(Integer)                  # 交易额 (正负值)
+    target_id = Column(String, nullable=True) # 关联的 item_id 或第三方支付宝流水号
+    
+    # 区块链级安全：哈希防篡改链
+    prev_hash = Column(String, nullable=True) # 上一笔交易的哈希值
+    tx_hash = Column(String)                  # 当前交易哈希值
+    
+    created_at = Column(DateTime, default=datetime.datetime.utcnow)

requirements.txt

@@ -4,4 +4,5 @@ pydantic
 huggingface_hub
 datasets
 python-multipart
-alibabacloud_dysmsapi20170525==2.0.24
+alibabacloud_dysmsapi20170525==2.0.24
+sqlalchemy

router_items.py

@@ -94,6 +94,10 @@ async def get_creators(sort: str = "downloads", limit: int = 20):
 
 @router.post("/api/items")
 async def create_item(item: ItemCreate):
+    # 【安全新增】：禁止负向消费防线
+    if item.price < 0:
+        raise HTTPException(status_code=400, detail="🚨 安全拦截：商品价格不能为负数")
+        
     items_db = db.load_data("items.json", default_data=[])
     new_item = {
         "id": f"{item.type}_{int(time.time())}_{uuid.uuid4().hex[:6]}", "type": item.type, "title": item.title, "author": item.author,
@@ -106,6 +110,10 @@ async def create_item(item: ItemCreate):
 
 @router.put("/api/items/{item_id}")
 async def update_item(item_id: str, update_data: ItemUpdate, author: str):
+    # 【安全新增】：禁止负向消费防线
+    if update_data.price is not None and update_data.price < 0:
+        raise HTTPException(status_code=400, detail="🚨 安全拦截：商品价格不能为负数")
+        
     items_db = db.load_data("items.json", default_data=[])
     for item in items_db:
         if item["id"] == item_id:

router_wallet.py

@@ -0,0 +1,124 @@
+# router_wallet.py
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+import time
+import uuid
+import hashlib
+from database_sql import get_db
+from models_sql import Wallet, Transaction, Ownership
+from models import RechargeRequest, WithdrawRequest, PurchaseRequest
+import 数据库连接 as json_db
+from router_users import VERIFY_CODES
+
+router = APIRouter()
+
+def calculate_tx_hash(tx_id, account, tx_type, amount, prev_hash):
+    """计算哈希值：依赖前一笔订单的哈希，实现链式防篡改"""
+    data = f"{tx_id}{account}{tx_type}{amount}{prev_hash}"
+    return hashlib.sha256(data.encode()).hexdigest()
+
+def record_transaction(db: Session, account: str, tx_type: str, amount: int, target_id: str = None):
+    """安全记录交易流水"""
+    last_tx = db.query(Transaction).filter(Transaction.account == account).order_by(Transaction.created_at.desc()).first()
+    prev_hash = last_tx.tx_hash if last_tx else "GENESIS"
+    
+    tx_id = f"TX_{int(time.time())}_{uuid.uuid4().hex[:8]}"
+    tx_hash = calculate_tx_hash(tx_id, account, tx_type, amount, prev_hash)
+    
+    new_tx = Transaction(tx_id=tx_id, account=account, tx_type=tx_type, amount=amount, target_id=target_id, prev_hash=prev_hash, tx_hash=tx_hash)
+    db.add(new_tx)
+    return new_tx
+
+@router.post("/api/wallet/recharge")
+async def recharge_points(req: RechargeRequest, db: Session = Depends(get_db)):
+    """充值接口"""
+    wallet = db.query(Wallet).filter(Wallet.account == req.account).first()
+    if not wallet:
+        wallet = Wallet(account=req.account)
+        db.add(wallet)
+    
+    # 模拟支付成功后直接给用户加钱
+    wallet.balance += req.amount
+    record_transaction(db, req.account, "RECHARGE", req.amount, "MOCK_ALIPAY_ORDER")
+    
+    db.commit()
+    return {"status": "success", "balance": wallet.balance}
+
+@router.post("/api/wallet/purchase")
+async def purchase_item(req: PurchaseRequest, db: Session = Depends(get_db)):
+    """消费购买接口：严密的并发防刷控制，以及早鸟保护机制"""
+    items_db = json_db.load_data("items.json", default_data=[])
+    item = next((i for i in items_db if i["id"] == req.item_id), None)
+    if not item: raise HTTPException(status_code=404, detail="商品不存在")
+    
+    price = int(item.get("price", 0))
+    author = item.get("author")
+    
+    # 1. 优先检查是否已经购买或获取过 (核心：无论后续怎么改价，只要所有权表里有你，直接放行)
+    owned = db.query(Ownership).filter(Ownership.account == req.account, Ownership.item_id == req.item_id).first()
+    if owned: return {"status": "success", "message": "已永久授权，无需重复扣费"}
+    
+    # ==========================================
+    # 2. 【核心修改】免费商品或创作者本人的“0元购”固化
+    # ==========================================
+    if price <= 0 or req.account == author:
+        # 直接写入所有权表，赋予永久权限
+        new_owner = Ownership(account=req.account, item_id=req.item_id)
+        db.add(new_owner)
+        # 记录一笔金额为 0 的流水，保证账本的完整溯源
+        record_transaction(db, req.account, "CONSUME", 0, req.item_id)
+        db.commit()
+        return {"status": "success", "message": "免费商品或创作者本人，已永久授权"}
+        
+    # 3. 开启数据库锁防止并发重复扣款 (排他锁)
+    buyer_wallet = db.query(Wallet).filter(Wallet.account == req.account).with_for_update().first()
+    if not buyer_wallet or buyer_wallet.balance < price:
+        raise HTTPException(status_code=400, detail="余额不足，请先充值")
+        
+    # 4. 执行扣费
+    buyer_wallet.balance -= price
+    record_transaction(db, req.account, "CONSUME", -price, req.item_id)
+    
+    # 5. 记录所有权
+    new_owner = Ownership(account=req.account, item_id=req.item_id)
+    db.add(new_owner)
+    
+    # 6. 卖家获得收益
+    author_wallet = db.query(Wallet).filter(Wallet.account == author).with_for_update().first()
+    if not author_wallet:
+        author_wallet = Wallet(account=author)
+        db.add(author_wallet)
+        
+    author_wallet.earn_balance += price
+    record_transaction(db, author, "EARN", price, req.item_id)
+    
+    db.commit()
+    return {"status": "success"}
+
+@router.post("/api/wallet/withdraw")
+async def withdraw_earnings(req: WithdrawRequest, db: Session = Depends(get_db)):
+    """提现接口"""
+    users_db = json_db.load_data("users.json", default_data={})
+    user = users_db.get(req.account)
+    if not user: raise HTTPException(status_code=404, detail="账号异常")
+    
+    cache_key = f"{user.get('email')}_withdraw"
+    cached = VERIFY_CODES.get(cache_key)
+    if not cached or cached["code"] != req.code or int(time.time()) > cached["expires_at"]:
+        raise HTTPException(status_code=400, detail="验证码��正确或已过期")
+        
+    wallet = db.query(Wallet).filter(Wallet.account == req.account).with_for_update().first()
+    if not wallet or wallet.earn_balance < req.amount:
+        raise HTTPException(status_code=400, detail="可提现收益不足")
+        
+    if req.amount < 100:
+        raise HTTPException(status_code=400, detail="最低提现金额为 100 积分")
+        
+    wallet.earn_balance -= req.amount
+    wallet.frozen_balance += req.amount
+    record_transaction(db, req.account, "WITHDRAW_FREEZE", -req.amount, req.alipay_account)
+    
+    VERIFY_CODES.pop(cache_key, None)
+    
+    db.commit()
+    return {"status": "success", "earn_balance": wallet.earn_balance}