KnowForge 2026.0.6：从知识管理走向会员、付费内容与 AI 服务

创建订单 -> 等待支付 -> 支付确认 -> 发放权益
                       -> 退款 -> 回收或调整权益
                       -> 关闭/过期

允许阅读 = 作者本人
        OR 管理员
        OR 拥有有效会员权益
        OR 拥有该内容的有效购买记录

python -m venv .venv
source .venv/bin/activate
pip install Flask==3.0.3

from flask import Flask, jsonify, request
from threading import Lock

app = Flask(__name__)
quota_lock = Lock()

# 演示数据：生产环境应改为数据库、真实登录态和订单记录。
USERS = {
    "alice": {"plan": "pro", "ai_remaining": 3, "purchases": set()},
    "bob": {"plan": "free", "ai_remaining": 0, "purchases": {"article-42"}},
}

ARTICLES = {
    "article-42": {
        "title": "构建团队知识库",
        "preview": "这是一段公开试读内容。",
        "body": "这是仅对会员或已购买用户开放的完整正文。",
        "paid": True,
    }
}


def current_user():
    # 仅用于演示。生产环境不要信任客户端提交的用户名或套餐字段。
    user_id = request.headers.get("X-Demo-User")
    return user_id, USERS.get(user_id)


def can_read(user_id, user, article_id):
    if not user:
        return False
    return user["plan"] == "pro" or article_id in user["purchases"]


@app.get("/articles/<article_id>")
def read_article(article_id):
    article = ARTICLES.get(article_id)
    if not article:
        return jsonify(error="article_not_found"), 404

    user_id, user = current_user()
    if article["paid"] and not can_read(user_id, user, article_id):
        return jsonify(
            title=article["title"],
            preview=article["preview"],
            locked=True,
            action="purchase_or_upgrade",
        ), 200

    return jsonify(**article, locked=False)


@app.post("/ai/ask")
def ask_ai():
    user_id, user = current_user()
    if not user:
        return jsonify(error="login_required"), 401

    question = (request.get_json(silent=True) or {}).get("question", "").strip()
    if not question:
        return jsonify(error="question_required"), 400

    # 锁只能保护单进程演示。生产环境应使用数据库原子更新或 Redis 脚本。
    with quota_lock:
        if user["ai_remaining"] <= 0:
            return jsonify(error="ai_quota_exhausted"), 429
        user["ai_remaining"] -= 1

    # 这里替换成实际模型或 KnowForge 支持的 AI 调用方式。
    answer = f"演示回答：已收到问题——{question}"
    return jsonify(answer=answer, ai_remaining=user["ai_remaining"])


if __name__ == "__main__":
    app.run(port=5000, debug=True)

python app.py

curl -s http://127.0.0.1:5000/articles/article-42 \
  -H 'X-Demo-User: unknown'

curl -s http://127.0.0.1:5000/articles/article-42 \
  -H 'X-Demo-User: bob'

curl -s -X POST http://127.0.0.1:5000/ai/ask \
  -H 'Content-Type: application/json' \
  -H 'X-Demo-User: alice' \
  -d '{"question":"如何整理项目复盘文档？"}'