feat: 新增一尘数据管理模块

- yichens_users: 用户表(积分/等级/商家认证)
- yichens_posts: 帖子表(分类/价格/置顶精华)
- yichens_replies: 回复表(楼层/点赞)
- yichens_follows: 关注关系表
- API路由: /api/v1/yichens/*

api/main.py

@@ -10,7 +10,7 @@ from fastapi import FastAPI, Request, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
-from api.routes import collections_router, prices_router, health_router
+from api.routes import collections_router, prices_router, health_router, yichens_router
 from api.middleware.response import ApiResponse
 
 # 配置日志
@@ -83,6 +83,7 @@ async def general_exception_handler(request: Request, exc: Exception):
 
 # 注册路由
 app.include_router(health_router)
+app.include_router(yichens_router)
 app.include_router(collections_router)
 app.include_router(prices_router)
 

api/routes/__init__.py

@@ -1,3 +1,4 @@
 from .collections import router as collections_router
 from .prices import router as prices_router
 from .health import router as health_router
+from .yichens import router as yichens_router

api/routes/yichens.py

@@ -0,0 +1,324 @@
+"""一尘数据管理路由"""
+from fastapi import APIRouter, HTTPException, Query
+from typing import Optional, List
+from datetime import datetime
+
+from api.middleware.response import ApiResponse, PaginatedData
+from database import db
+
+router = APIRouter(prefix="/api/v1/yichens", tags=["yichens"])
+
+# ============ 用户管理 ============
+
+@router.get("/users", response_model=ApiResponse)
+async def list_yichens_users(
+    username: Optional[str] = Query(None, description="用户名搜索"),
+    is_seller: Optional[bool] = Query(None, description="是否商家"),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100)
+):
+    """获取一尘用户列表"""
+    offset = (page - 1) * page_size
+    where_clauses = []
+    params = []
+    
+    if username:
+        where_clauses.append("username LIKE %s")
+        params.append(f"%{username}%")
+    if is_seller is not None:
+        where_clauses.append("is_seller = %s")
+        params.append(is_seller)
+    
+    where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
+    
+    with db.get_cursor() as cursor:
+        cursor.execute(f"SELECT COUNT(*) as total FROM yichens_users{where_sql}", params)
+        total = cursor.fetchone()["total"]
+        
+        cursor.execute(f"""
+            SELECT * FROM yichens_users{where_sql}
+            ORDER BY crawl_latest_at DESC LIMIT %s OFFSET %s
+        """, params + [page_size, offset])
+        items = cursor.fetchall()
+    
+    return ApiResponse.success(PaginatedData.create(items, page, page_size, total))
+
+@router.get("/users/{user_id}", response_model=ApiResponse)
+async def get_yichens_user(user_id: str):
+    """获取一尘用户详情"""
+    with db.get_cursor() as cursor:
+        cursor.execute("SELECT * FROM yichens_users WHERE user_id = %s", (user_id,))
+        item = cursor.fetchone()
+    
+    if not item:
+        raise HTTPException(status_code=404, detail="User not found")
+    
+    return ApiResponse.success(item)
+
+@router.post("/users", response_model=ApiResponse, status_code=201)
+async def create_yichens_user(user_data: dict):
+    """创建/更新一尘用户"""
+    required = ["user_id", "username"]
+    for field in required:
+        if field not in user_data:
+            raise HTTPException(status_code=400, detail=f"Missing required field: {field}")
+    
+    with db.get_cursor() as cursor:
+        cursor.execute("""
+            INSERT INTO yichens_users (
+                user_id, username, nickname, avatar_url, user_level,
+                credit_score, register_date, last_active_at, is_seller,
+                seller_rating, is_verified, bio, province
+            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+            ON DUPLICATE KEY UPDATE
+                username = VALUES(username),
+                nickname = VALUES(nickname),
+                avatar_url = VALUES(avatar_url),
+                user_level = VALUES(user_level),
+                credit_score = VALUES(credit_score),
+                is_seller = VALUES(is_seller),
+                seller_rating = VALUES(seller_rating),
+                is_verified = VALUES(is_verified),
+                bio = VALUES(bio),
+                province = VALUES(province),
+                crawl_latest_at = NOW()
+        """, (
+            user_data.get("user_id"),
+            user_data.get("username"),
+            user_data.get("nickname"),
+            user_data.get("avatar_url"),
+            user_data.get("user_level"),
+            user_data.get("credit_score", 0),
+            user_data.get("register_date"),
+            user_data.get("last_active_at"),
+            user_data.get("is_seller", False),
+            user_data.get("seller_rating"),
+            user_data.get("is_verified", False),
+            user_data.get("bio"),
+            user_data.get("province")
+        ))
+    
+    return ApiResponse.success({"user_id": user_data["user_id"]}, "User saved successfully")
+
+# ============ 帖子管理 ============
+
+@router.get("/posts", response_model=ApiResponse)
+async def list_yichens_posts(
+    category: Optional[str] = Query(None, description="分类筛选"),
+    post_type: Optional[str] = Query(None, description="帖子类型"),
+    author_username: Optional[str] = Query(None, description="作者用户名"),
+    keyword: Optional[str] = Query(None, description="关键词搜索"),
+    min_price: Optional[float] = Query(None, description="最低价格"),
+    max_price: Optional[float] = Query(None, description="最高价格"),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100)
+):
+    """获取一尘帖子列表"""
+    offset = (page - 1) * page_size
+    where_clauses = []
+    params = []
+    
+    if category:
+        where_clauses.append("category = %s")
+        params.append(category)
+    if post_type:
+        where_clauses.append("post_type = %s")
+        params.append(post_type)
+    if author_username:
+        where_clauses.append("author_username LIKE %s")
+        params.append(f"%{author_username}%")
+    if keyword:
+        where_clauses.append("MATCH(title, content) AGAINST(%s IN NATURAL LANGUAGE MODE)")
+        params.append(keyword)
+    if min_price is not None:
+        where_clauses.append("price >= %s")
+        params.append(min_price)
+    if max_price is not None:
+        where_clauses.append("price <= %s")
+        params.append(max_price)
+    
+    where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
+    
+    with db.get_cursor() as cursor:
+        cursor.execute(f"SELECT COUNT(*) as total FROM yichens_posts{where_sql}", params)
+        total = cursor.fetchone()["total"]
+        
+        cursor.execute(f"""
+            SELECT * FROM yichens_posts{where_sql}
+            ORDER BY created_at DESC LIMIT %s OFFSET %s
+        """, params + [page_size, offset])
+        items = cursor.fetchall()
+    
+    return ApiResponse.success(PaginatedData.create(items, page, page_size, total))
+
+@router.get("/posts/{post_id}", response_model=ApiResponse)
+async def get_yichens_post(post_id: str):
+    """获取一尘帖子详情"""
+    with db.get_cursor() as cursor:
+        cursor.execute("SELECT * FROM yichens_posts WHERE post_id = %s", (post_id,))
+        item = cursor.fetchone()
+    
+    if not item:
+        raise HTTPException(status_code=404, detail="Post not found")
+    
+    return ApiResponse.success(item)
+
+@router.post("/posts", response_model=ApiResponse, status_code=201)
+async def create_yichens_post(post_data: dict):
+    """创建/更新一尘帖子"""
+    required = ["post_id", "topic_id", "title"]
+    for field in required:
+        if field not in post_data:
+            raise HTTPException(status_code=400, detail=f"Missing required field: {field}")
+    
+    with db.get_cursor() as cursor:
+        cursor.execute("""
+            INSERT INTO yichens_posts (
+                post_id, topic_id, title, content, content_html,
+                author_id, author_username, category, sub_category,
+                post_type, price, price_unit, view_count, reply_count,
+                like_count, is_top, is_essence, is_closed, created_at, updated_at
+            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+            ON DUPLICATE KEY UPDATE
+                title = VALUES(title),
+                content = VALUES(content),
+                content_html = VALUES(content_html),
+                author_username = VALUES(author_username),
+                category = VALUES(category),
+                post_type = VALUES(post_type),
+                price = VALUES(price),
+                view_count = VALUES(view_count),
+                reply_count = VALUES(reply_count),
+                like_count = VALUES(like_count),
+                is_top = VALUES(is_top),
+                is_essence = VALUES(is_essence),
+                is_closed = VALUES(is_closed),
+                updated_at = VALUES(updated_at),
+                crawled_at = NOW()
+        """, (
+            post_data.get("post_id"),
+            post_data.get("topic_id"),
+            post_data.get("title"),
+            post_data.get("content"),
+            post_data.get("content_html"),
+            post_data.get("author_id"),
+            post_data.get("author_username"),
+            post_data.get("category"),
+            post_data.get("sub_category"),
+            post_data.get("post_type", "normal"),
+            post_data.get("price"),
+            post_data.get("price_unit"),
+            post_data.get("view_count", 0),
+            post_data.get("reply_count", 0),
+            post_data.get("like_count", 0),
+            post_data.get("is_top", False),
+            post_data.get("is_essence", False),
+            post_data.get("is_closed", False),
+            post_data.get("created_at"),
+            post_data.get("updated_at")
+        ))
+    
+    return ApiResponse.success({"post_id": post_data["post_id"]}, "Post saved successfully")
+
+# ============ 回复管理 ============
+
+@router.get("/posts/{post_id}/replies", response_model=ApiResponse)
+async def get_post_replies(
+    post_id: str,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=200)
+):
+    """获取帖子回复列表"""
+    offset = (page - 1) * page_size
+    
+    with db.get_cursor() as cursor:
+        cursor.execute(
+            "SELECT COUNT(*) as total FROM yichens_replies WHERE post_id = %s",
+            (post_id,)
+        )
+        total = cursor.fetchone()["total"]
+        
+        cursor.execute("""
+            SELECT * FROM yichens_replies
+            WHERE post_id = %s
+            ORDER BY floor_number ASC LIMIT %s OFFSET %s
+        """, (post_id, page_size, offset))
+        items = cursor.fetchall()
+    
+    return ApiResponse.success(PaginatedData.create(items, page, page_size, total))
+
+@router.post("/replies", response_model=ApiResponse, status_code=201)
+async def create_yichens_reply(reply_data: dict):
+    """创建/更新回复"""
+    required = ["reply_id", "post_id"]
+    for field in required:
+        if field not in reply_data:
+            raise HTTPException(status_code=400, detail=f"Missing required field: {field}")
+    
+    with db.get_cursor() as cursor:
+        cursor.execute("""
+            INSERT INTO yichens_replies (
+                reply_id, post_id, floor_number, content, content_html,
+                author_id, author_username, like_count, created_at
+            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
+            ON DUPLICATE KEY UPDATE
+                content = VALUES(content),
+                content_html = VALUES(content_html),
+                like_count = VALUES(like_count),
+                crawled_at = NOW()
+        """, (
+            reply_data.get("reply_id"),
+            reply_data.get("post_id"),
+            reply_data.get("floor_number"),
+            reply_data.get("content"),
+            reply_data.get("content_html"),
+            reply_data.get("author_id"),
+            reply_data.get("author_username"),
+            reply_data.get("like_count", 0),
+            reply_data.get("created_at")
+        ))
+    
+    return ApiResponse.success({"reply_id": reply_data["reply_id"]}, "Reply saved successfully")
+
+# ============ 统计 ============
+
+@router.get("/statistics", response_model=ApiResponse)
+async def get_yichens_statistics():
+    """获取一尘数据统计"""
+    with db.get_cursor() as cursor:
+        cursor.execute("SELECT COUNT(*) as total FROM yichens_users")
+        total_users = cursor.fetchone()["total"]
+        
+        cursor.execute("SELECT COUNT(*) as total FROM yichens_posts")
+        total_posts = cursor.fetchone()["total"]
+        
+        cursor.execute("""
+            SELECT COUNT(*) as total FROM yichens_users WHERE is_seller = TRUE
+        """)
+        seller_count = cursor.fetchone()["total"]
+        
+        cursor.execute("""
+            SELECT category, COUNT(*) as count 
+            FROM yichens_posts 
+            WHERE category IS NOT NULL
+            GROUP BY category
+            ORDER BY count DESC LIMIT 10
+        """)
+        category_stats = cursor.fetchall()
+        
+        cursor.execute("""
+            SELECT DATE(crawled_at) as date, COUNT(*) as count
+            FROM yichens_posts
+            WHERE crawled_at >= DATE_SUB(NOW(), INTERVAL 7 DAY)
+            GROUP BY DATE(crawled_at)
+            ORDER BY date DESC
+        """)
+        daily_stats = cursor.fetchall()
+    
+    return ApiResponse.success({
+        "total_users": total_users,
+        "total_posts": total_posts,
+        "seller_count": seller_count,
+        "category_stats": category_stats,
+        "daily_stats": daily_stats
+    })