"""一尘数据管理路由""" 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, get_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=500) ): """获取一尘用户列表""" 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()[0] 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="最高价格"), time_range: Optional[str] = Query(None, description="时间范围: 1h, 6h, 24h, today"), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=500) ): offset = (page - 1) * page_size where_clauses = ["1=1"] params = [] if category: where_clauses.append("category = %s") params.append(category) if post_type == 'other': where_clauses.append("post_type NOT IN ('deal','want')") elif 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("(title ILIKE %s OR content ILIKE %s)") params.extend([f"%{keyword}%", f"%{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) if time_range == '1h': where_clauses.append("post_time >= NOW() - INTERVAL '1 hour'") elif time_range == '6h': where_clauses.append("post_time >= NOW() - INTERVAL '6 hours'") elif time_range == '24h': where_clauses.append("post_time >= NOW() - INTERVAL '24 hours'") elif time_range == 'today': where_clauses.append("post_time::date = CURRENT_DATE") where_sql = " WHERE " + " AND ".join(where_clauses) with get_db() as conn: cur = conn.cursor() cur.execute(f"SELECT COUNT(*) as total FROM yichens_posts{where_sql}", params) total = cur.fetchone()[0] cur.execute(f"SELECT * FROM yichens_posts{where_sql} ORDER BY post_time DESC LIMIT %s OFFSET %s", params + [page_size, offset]) rows = cur.fetchall() cols = [desc[0] for desc in cur.description] items = [dict(zip(cols, row)) for row in rows] cur.close() 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()[0] 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 >= NOW() - INTERVAL '7 days' GROUP BY DATE(crawled_at) ORDER BY date DESC """) daily_stats = cursor.fetchall() # Add today counts cursor.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_time::date = CURRENT_DATE") today_count = cursor.fetchone()["count"] cursor.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'deal' AND post_time::date = CURRENT_DATE") deal_count = cursor.fetchone()["count"] cursor.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'want' AND post_time::date = CURRENT_DATE") want_count = cursor.fetchone()["count"] cursor.execute("SELECT COUNT(*) FROM yichens_posts WHERE has_lifebuoy = TRUE AND post_time::date = CURRENT_DATE") lifebuoy_count = cursor.fetchone()["count"] return ApiResponse.success({ "total_users": total_users, "total_posts": total_posts, "today_count": today_count, "deal_count": deal_count, "want_count": want_count, "lifebuoy_count": lifebuoy_count, "seller_count": seller_count, "category_stats": category_stats, "daily_stats": daily_stats }) @router.get("/report/categories", response_model=ApiResponse) async def get_category_report(): with get_db() as conn: cur = conn.cursor() cur.execute(""" SELECT COALESCE(category, '其他') as category, COUNT(*) as cnt FROM yichens_posts WHERE post_time::date = CURRENT_DATE GROUP BY COALESCE(category, '其他') ORDER BY cnt DESC """) today_cats = [{"category": r[0], "count": r[1]} for r in cur.fetchall()] cur.execute(""" SELECT COALESCE(category, '其他') as category, COUNT(*) as cnt FROM yichens_posts WHERE post_time >= NOW() - INTERVAL '1 hour' GROUP BY COALESCE(category, '其他') ORDER BY cnt DESC """) last1h_cats = [{"category": r[0], "count": r[1]} for r in cur.fetchall()] cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_time::date = CURRENT_DATE") today_total = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_time >= NOW() - INTERVAL '1 hour'") last1h_total = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'deal' AND post_time::date = CURRENT_DATE") today_deal = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'want' AND post_time::date = CURRENT_DATE") today_want = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE has_lifebuoy = TRUE AND post_time::date = CURRENT_DATE") today_lifebuoy = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'deal' AND post_time >= NOW() - INTERVAL '1 hour'") last1h_deal = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'want' AND post_time >= NOW() - INTERVAL '1 hour'") last1h_want = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE has_lifebuoy = TRUE AND post_time >= NOW() - INTERVAL '1 hour'") last1h_lifebuoy = cur.fetchone()[0] cur.close() return ApiResponse.success({ "today": {"total": today_total, "deal": today_deal, "want": today_want, "lifebuoy": today_lifebuoy, "categories": today_cats}, "last_1h": {"total": last1h_total, "deal": last1h_deal, "want": last1h_want, "lifebuoy": last1h_lifebuoy, "categories": last1h_cats} }) @router.get("/report/stats", response_model=ApiResponse) async def get_longtian_stats(): """龙钞统计分析""" with get_db() as conn: cur = conn.cursor() # 收购关键词 want_kw = '%(龙|龙钞|小龙钞)%' want_cond = "(title ILIKE '%龙%' OR title ILIKE '%龙钞%' OR title ILIKE '%小龙钞%') AND (title ILIKE '%收%' OR title ILIKE '%求%' OR title ILIKE '%购%')" # 出售关键词 deal_kw = '%(龙|龙钞|小龙钞)%' deal_cond = "(title ILIKE '%龙%' OR title ILIKE '%龙钞%' OR title ILIKE '%小龙钞%') AND (title ILIKE '%出%' OR title ILIKE '%售%')" def count_by_pattern(cur, base_cond, extra_kw): """统计满足基本条件+特殊关键词的帖子数量""" where = base_cond + " AND " + extra_kw + " AND post_time::date = CURRENT_DATE" cur.execute(f"SELECT COUNT(*) FROM yichens_posts WHERE {where}") return cur.fetchone()[0] def avg_price_by_pattern(cur, base_cond, extra_kw): """统计满足基本条件+特殊关键词的帖子平均价格""" where = base_cond + " AND " + extra_kw + " AND post_time::date = CURRENT_DATE AND price IS NOT NULL AND price > 0" cur.execute(f"SELECT COALESCE(AVG(price), 0), COUNT(*) FROM yichens_posts WHERE {where}") r = cur.fetchone() return {"avg": int(r[0]) if r[0] else 0, "count": r[1]} # 收购类 - 带4标十 c1 = count_by_pattern(cur, want_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'") p1 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'") # 收购类 - 无47标十 c2 = count_by_pattern(cur, want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'") p2 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'") # 收购类 - 无247标十 c3 = count_by_pattern(cur, want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'") p3 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'") # 收购类 - 无247标百 c4 = count_by_pattern(cur, want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')") p4 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')") # 收购类 - 无47标百 c5 = count_by_pattern(cur, want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')") p5 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')") # 出售类 - 带4标十 c6 = count_by_pattern(cur, deal_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'") p6 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'") # 出售类 - 无47标十 c7 = count_by_pattern(cur, deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'") p7 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'") # 出售类 - 无247标十 c8 = count_by_pattern(cur, deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'") p8 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'") # 出售类 - 无247标百 c9 = count_by_pattern(cur, deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')") p9 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')") # 出售类 - 无47标百 c10 = count_by_pattern(cur, deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')") p10 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')") cur.close() return ApiResponse.success({ "want": { "dai4_biao10": {"count": c1, **p1}, "wu47_biao10": {"count": c2, **p2}, "wu247_biao10": {"count": c3, **p3}, "wu247_biaobai": {"count": c4, **p4}, "wu47_biaobai": {"count": c5, **p5}, }, "deal": { "dai4_biao10": {"count": c6, **p6}, "wu47_biao10": {"count": c7, **p7}, "wu247_biao10": {"count": c8, **p8}, "wu247_biaobai": {"count": c9, **p9}, "wu47_biaobai": {"count": c10, **p10}, } }) @router.get("/report/posts", response_model=ApiResponse) async def get_report_posts( stat_type: str = Query(..., description="统计类型: want_dai4_biao10, want_wu47_biao10, deal_dai4_biao10, etc"), page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=500) ): """获取统计报表对应的帖子列表""" offset = (page - 1) * page_size # 基础条件:龙钞类 longtian_cond = "(title ILIKE '%龙%' OR title ILIKE '%龙钞%' OR title ILIKE '%小龙钞%')" # 收购/出售条件 want_cond = longtian_cond + " AND (title ILIKE '%收%' OR title ILIKE '%求%' OR title ILIKE '%购%')" deal_cond = longtian_cond + " AND (title ILIKE '%出%' OR title ILIKE '%售%')" # stat_type 到 WHERE 条件的映射 stat_map = { "want_dai4_biao10": (want_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'"), "want_wu47_biao10": (want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'"), "want_wu247_biao10": (want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'"), "want_wu247_biaobai": (want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')"), "want_wu47_biaobai": (want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')"), "deal_dai4_biao10": (deal_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'"), "deal_wu47_biao10": (deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'"), "deal_wu247_biao10": (deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'"), "deal_wu247_biaobai": (deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')"), "deal_wu47_biaobai": (deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')"), } if stat_type not in stat_map: return ApiResponse.success({"items": [], "total": 0, "page": page, "page_size": page_size}) base_cond, extra_cond = stat_map[stat_type] where_sql = f" WHERE {base_cond} AND {extra_cond} AND post_time::date = CURRENT_DATE" with get_db() as conn: cur = conn.cursor() cur.execute(f"SELECT COUNT(*) FROM yichens_posts{where_sql}") total = cur.fetchone()[0] cur.execute(f"SELECT * FROM yichens_posts{where_sql} ORDER BY post_time DESC LIMIT {page_size} OFFSET {offset}") rows = cur.fetchall() cols = [desc[0] for desc in cur.description] items = [dict(zip(cols, row)) for row in rows] cur.close() return ApiResponse.success({"items": items, "total": total, "page": page, "page_size": page_size})