refactor: 重构 API 架构，规范化命名和响应格式

- 统一响应格式: {code, message, data, timestamp}
- RESTful 路由: /api/v1/collections, /api/v1/prices
- 分页封装: PaginatedData
- 路由模块化: routes/collections.py, routes/prices.py, routes/health.py
- 数据模型: Pydantic models for request/response
- 数据库连接池
- 配置文件: config/config.yaml
- 爬虫模块: crawlers/yichens_spider.py

api/__init__.py

@@ -0,0 +1,2 @@
+"""CoolBotDataSys API Module"""
+__version__ = "1.0.0"

api/main.py

@@ -0,0 +1,167 @@
+"""
+CoolBotDataSys API 服务
+酷博特数据分析系统 RESTful API
+
+版本: 1.0.0
+"""
+import logging
+from contextlib import asynccontextmanager
+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.middleware.response import ApiResponse
+
+# 配置日志
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """应用生命周期管理"""
+    logger.info("🚀 CoolBotDataSys API 启动...")
+    yield
+    logger.info("👋 CoolBotDataSys API 关闭...")
+
+# 创建 FastAPI 应用
+app = FastAPI(
+    title="CoolBotDataSys API",
+    description="""
+## CoolBotDataSys - 酷博特数据分析系统
+
+### 功能模块
+- **藏品管理** - 藏品的 CRUD 操作
+- **价格追踪** - 历史价格查询和趋势分析
+- **爬虫调度** - 数据采集任务管理
+- **统计报表** - 数据统计和报表生成
+
+### 认证方式
+当前版本暂不需要认证，后续添加 API Key 认证。
+    """,
+    version="1.0.0",
+    lifespan=lifespan,
+    docs_url="/docs",
+    redoc_url="/redoc"
+)
+
+# CORS 配置
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 全局异常处理
+@app.exception_handler(HTTPException)
+async def http_exception_handler(request: Request, exc: HTTPException):
+    """HTTP 异常处理"""
+    return JSONResponse(
+        status_code=exc.status_code,
+        content=ApiResponse.error(
+            code=exc.status_code * 100,
+            message=exc.detail
+        ).model_dump()
+    )
+
+@app.exception_handler(Exception)
+async def general_exception_handler(request: Request, exc: Exception):
+    """通用异常处理"""
+    logger.error(f"未处理异常: {exc}", exc_info=True)
+    return JSONResponse(
+        status_code=500,
+        content=ApiResponse.error(
+            code=50000,
+            message="Internal server error"
+        ).model_dump()
+    )
+
+# 注册路由
+app.include_router(health_router)
+app.include_router(collections_router)
+app.include_router(prices_router)
+
+# 统计接口
+from api.middleware.response import ApiResponse
+from database import db
+
+@app.get("/api/v1/statistics", response_model=ApiResponse, tags=["statistics"])
+async def get_statistics():
+    """获取系统统计信息"""
+    with db.get_cursor() as cursor:
+        # 藏品总数
+        cursor.execute("SELECT COUNT(*) as cnt FROM collections")
+        total_collections = cursor.fetchone()["cnt"]
+        
+        # 价格记录总数
+        cursor.execute("SELECT COUNT(*) as cnt FROM price_history")
+        total_price_records = cursor.fetchone()["cnt"]
+        
+        # 今日新增价格记录
+        cursor.execute("""
+            SELECT COUNT(*) as cnt FROM price_history 
+            WHERE DATE(crawled_at) = CURDATE()
+        """)
+        today_price_records = cursor.fetchone()["cnt"]
+        
+        # 最近爬虫运行时间
+        cursor.execute("""
+            SELECT MAX(finished_at) as last_run 
+            FROM crawl_logs WHERE status = 'success'
+        """)
+        last_crawl = cursor.fetchone()["last_run"]
+        
+        # 分类统计
+        cursor.execute("""
+            SELECT category, COUNT(*) as count 
+            FROM collections 
+            WHERE category IS NOT NULL
+            GROUP BY category
+        """)
+        category_stats = {row["category"]: row["count"] for row in cursor.fetchall()}
+    
+    return ApiResponse.success({
+        "total_collections": total_collections,
+        "total_price_records": total_price_records,
+        "today_price_records": today_price_records,
+        "last_crawl_time": last_crawl,
+        "category_stats": category_stats
+    })
+
+# 爬虫调度接口
+from crawlers.yichens_spider import YichensSpider
+
+@app.post("/api/v1/crawl-jobs/trigger", response_model=ApiResponse, tags=["crawl"])
+async def trigger_crawl(source: str = "yichens"):
+    """触发爬虫任务"""
+    if source == "yichens":
+        spider = YichensSpider()
+        items = spider.run()
+        return ApiResponse.success({
+            "source": source,
+            "items_collected": len(items),
+            "status": "completed"
+        }, "Crawl job triggered successfully")
+    else:
+        raise HTTPException(status_code=400, detail=f"Unknown source: {source}")
+
+@app.get("/api/v1/crawl-jobs", response_model=ApiResponse, tags=["crawl"])
+async def get_crawl_jobs(limit: int = 20):
+    """获取爬虫运行历史"""
+    with db.get_cursor() as cursor:
+        cursor.execute("""
+            SELECT * FROM crawl_logs 
+            ORDER BY finished_at DESC LIMIT %s
+        """, (limit,))
+        items = cursor.fetchall()
+    return ApiResponse.success(items)
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8080)

api/middleware/response.py

@@ -0,0 +1,47 @@
+"""统一响应格式封装"""
+from typing import Any, Optional, Generic, TypeVar
+from pydantic import BaseModel
+from datetime import datetime
+
+T = TypeVar("T")
+
+class ApiResponse(BaseModel, Generic[T]):
+    """统一 API 响应格式"""
+    code: int = 0
+    message: str = "success"
+    data: Optional[T] = None
+    timestamp: str = ""
+    
+    def __init__(self, code: int = 0, message: str = "success", data: T = None, **kwargs):
+        super().__init__(
+            code=code,
+            message=message,
+            data=data,
+            timestamp=datetime.now().isoformat() + "Z"
+        )
+    
+    @classmethod
+    def success(cls, data: T = None, message: str = "success"):
+        return cls(code=0, message=message, data=data)
+    
+    @classmethod
+    def error(cls, code: int = 50000, message: str = "Internal error"):
+        return cls(code=code, message=message, data=None)
+
+class PaginatedData(BaseModel, Generic[T]):
+    """分页数据封装"""
+    items: list[T]
+    pagination: dict
+    
+    @classmethod
+    def create(cls, items: list, page: int = 1, page_size: int = 20, total: int = 0):
+        total_pages = (total + page_size - 1) // page_size if page_size > 0 else 0
+        return cls(
+            items=items,
+            pagination={
+                "page": page,
+                "page_size": page_size,
+                "total": total,
+                "total_pages": total_pages
+            }
+        )

api/models/__init__.py

@@ -0,0 +1,2 @@
+from .request import CollectionCreate, CollectionUpdate, PriceQuery
+from .response import CollectionResponse, PriceHistoryResponse, CrawlLogResponse, StatisticsResponse

api/models/request.py

@@ -0,0 +1,43 @@
+"""API 请求模型"""
+from pydantic import BaseModel, Field
+from typing import Optional
+from datetime import datetime
+
+class CollectionCreate(BaseModel):
+    """创建藏品请求"""
+    name: str = Field(..., description="藏品名称", min_length=1, max_length=255)
+    category: Optional[str] = Field(None, description="品类（龙钞/蛇钞/马钞等）")
+    serial_number: Optional[str] = Field(None, description="号码（唯一标识）", max_length=100)
+    rarity: str = Field("通货", description="珍惜度（通货/标十/标百/狮子号）")
+    ownership_type: str = Field("自持", description="持仓类型（自持/代存）")
+    is_graded: bool = Field(False, description="是否评级")
+    grading_company: Optional[str] = Field(None, description="评级公司（PCGS/NGC/爱藏）")
+    grading_score: Optional[str] = Field(None, description="评级分数（如68/69/70）")
+    cost_price: Optional[float] = Field(None, description="成本价（元）")
+    current_price: Optional[float] = Field(None, description="当前市场价（元）")
+    status: str = Field("in_collection", description="状态（in_collection/sold/lost）")
+    remark: Optional[str] = Field(None, description="备注")
+
+class CollectionUpdate(BaseModel):
+    """更新藏品请求"""
+    name: Optional[str] = Field(None, description="藏品名称")
+    category: Optional[str] = Field(None, description="品类")
+    serial_number: Optional[str] = Field(None, description="号码")
+    rarity: Optional[str] = Field(None, description="珍惜度")
+    ownership_type: Optional[str] = Field(None, description="持仓类型")
+    is_graded: Optional[bool] = Field(None, description="是否评级")
+    grading_company: Optional[str] = Field(None, description="评级公司")
+    grading_score: Optional[str] = Field(None, description="评级分数")
+    cost_price: Optional[float] = Field(None, description="成本价")
+    current_price: Optional[float] = Field(None, description="当前市场价")
+    status: Optional[str] = Field(None, description="状态")
+    remark: Optional[str] = Field(None, description="备注")
+
+class PriceQuery(BaseModel):
+    """价格查询参数"""
+    collection_id: Optional[int] = Field(None, description="藏品ID")
+    source: Optional[str] = Field(None, description="数据来源（yichens/aicang/douyin）")
+    start_date: Optional[datetime] = Field(None, description="开始时间")
+    end_date: Optional[datetime] = Field(None, description="结束时间")
+    page: int = Field(1, ge=1, description="页码")
+    page_size: int = Field(20, ge=1, le=100, description="每页数量")

api/models/response.py

@@ -0,0 +1,51 @@
+"""API 响应模型"""
+from pydantic import BaseModel
+from typing import Optional, List, Any
+from datetime import datetime
+
+class CollectionResponse(BaseModel):
+    """藏品响应"""
+    id: int
+    name: str
+    category: Optional[str] = None
+    serial_number: Optional[str] = None
+    rarity: str = "通货"
+    ownership_type: str = "自持"
+    is_graded: bool = False
+    grading_company: Optional[str] = None
+    grading_score: Optional[str] = None
+    cost_price: Optional[float] = None
+    current_price: Optional[float] = None
+    status: str = "in_collection"
+    remark: Optional[str] = None
+    created_at: datetime
+    updated_at: datetime
+
+class PriceHistoryResponse(BaseModel):
+    """价格历史响应"""
+    id: int
+    collection_id: Optional[int] = None
+    source: str
+    price: float
+    price_unit: str = "元/张"
+    price_type: str = "挂牌价"
+    url: Optional[str] = None
+    crawled_at: datetime
+
+class CrawlLogResponse(BaseModel):
+    """爬虫日志响应"""
+    id: int
+    source: str
+    status: str
+    items_count: int = 0
+    error_message: Optional[str] = None
+    started_at: Optional[datetime] = None
+    finished_at: Optional[datetime] = None
+
+class StatisticsResponse(BaseModel):
+    """统计信息响应"""
+    total_collections: int = 0
+    total_price_records: int = 0
+    today_price_records: int = 0
+    last_crawl_time: Optional[datetime] = None
+    category_stats: dict = {}

api/routes/__init__.py

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

api/routes/collections.py

@@ -0,0 +1,156 @@
+"""藏品路由"""
+from fastapi import APIRouter, HTTPException, Query
+from typing import Optional, List
+from datetime import datetime
+
+from api.models.request import CollectionCreate, CollectionUpdate
+from api.models.response import CollectionResponse
+from api.middleware.response import ApiResponse, PaginatedData
+from database import db
+
+router = APIRouter(prefix="/api/v1/collections", tags=["collections"])
+
+@router.get("", response_model=ApiResponse)
+async def list_collections(
+    category: Optional[str] = Query(None, description="品类筛选"),
+    rarity: Optional[str] = Query(None, description="珍惜度筛选"),
+    status: Optional[str] = Query(None, description="状态筛选"),
+    page: int = Query(1, ge=1, description="页码"),
+    page_size: int = Query(20, ge=1, le=100, description="每页数量")
+):
+    """获取藏品列表（分页）"""
+    offset = (page - 1) * page_size
+    
+    where_clauses = []
+    params = []
+    if category:
+        where_clauses.append("category = %s")
+        params.append(category)
+    if rarity:
+        where_clauses.append("rarity = %s")
+        params.append(rarity)
+    if status:
+        where_clauses.append("status = %s")
+        params.append(status)
+    
+    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 collections{where_sql}", params)
+        total = cursor.fetchone()["total"]
+        
+        query_sql = f"""
+            SELECT * FROM collections{where_sql}
+            ORDER BY updated_at DESC LIMIT %s OFFSET %s
+        """
+        cursor.execute(query_sql, params + [page_size, offset])
+        items = cursor.fetchall()
+    
+    return ApiResponse.success(
+        PaginatedData.create(items, page, page_size, total)
+    )
+
+@router.get("/{collection_id}", response_model=ApiResponse)
+async def get_collection(collection_id: int):
+    """获取单个藏品详情"""
+    with db.get_cursor() as cursor:
+        cursor.execute("SELECT * FROM collections WHERE id = %s", (collection_id,))
+        item = cursor.fetchone()
+    
+    if not item:
+        raise HTTPException(status_code=404, detail="Collection not found")
+    
+    return ApiResponse.success(item)
+
+@router.post("", response_model=ApiResponse, status_code=201)
+async def create_collection(collection: CollectionCreate):
+    """创建新藏品"""
+    with db.get_cursor() as cursor:
+        cursor.execute("""
+            INSERT INTO collections (
+                name, category, serial_number, rarity, ownership_type,
+                is_graded, grading_company, grading_score,
+                cost_price, current_price, status, remark
+            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+        """, (
+            collection.name, collection.category, collection.serial_number,
+            collection.rarity, collection.ownership_type, collection.is_graded,
+            collection.grading_company, collection.grading_score,
+            collection.cost_price, collection.current_price, collection.status, collection.remark
+        ))
+        
+        new_id = cursor.lastrowid
+        cursor.execute("SELECT * FROM collections WHERE id = %s", (new_id,))
+        item = cursor.fetchone()
+    
+    return ApiResponse.success(item, "Collection created successfully")
+
+@router.put("/{collection_id}", response_model=ApiResponse)
+async def update_collection(collection_id: int, collection: CollectionUpdate):
+    """更新藏品"""
+    update_fields = []
+    params = []
+    
+    for field, value in collection.model_dump(exclude_unset=True).items():
+        if value is not None:
+            update_fields.append(f"{field} = %s")
+            params.append(value)
+    
+    if not update_fields:
+        raise HTTPException(status_code=400, detail="No fields to update")
+    
+    params.append(collection_id)
+    set_clause = ", ".join(update_fields)
+    
+    with db.get_cursor() as cursor:
+        sql = f"UPDATE collections SET {set_clause}, updated_at = NOW() WHERE id = %s"
+        cursor.execute(sql, params)
+        
+        if cursor.rowcount == 0:
+            raise HTTPException(status_code=404, detail="Collection not found")
+        
+        cursor.execute("SELECT * FROM collections WHERE id = %s", (collection_id,))
+        item = cursor.fetchone()
+    
+    return ApiResponse.success(item, "Collection updated successfully")
+
+@router.delete("/{collection_id}", response_model=ApiResponse)
+async def delete_collection(collection_id: int):
+    """删除藏品"""
+    with db.get_cursor() as cursor:
+        cursor.execute("DELETE FROM collections WHERE id = %s", (collection_id,))
+        if cursor.rowcount == 0:
+            raise HTTPException(status_code=404, detail="Collection not found")
+    
+    return ApiResponse.success(message="Collection deleted successfully")
+
+@router.get("/{collection_id}/prices", response_model=ApiResponse)
+async def get_collection_prices(
+    collection_id: int,
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100)
+):
+    """获取藏品的价柗历史"""
+    offset = (page - 1) * page_size
+    
+    with db.get_cursor() as cursor:
+        cursor.execute("SELECT id FROM collections WHERE id = %s", (collection_id,))
+        if not cursor.fetchone():
+            raise HTTPException(status_code=404, detail="Collection not found")
+        
+        cursor.execute(
+            "SELECT COUNT(*) as total FROM price_history WHERE collection_id = %s",
+            (collection_id,)
+        )
+        total = cursor.fetchone()["total"]
+        
+        cursor.execute("""
+            SELECT * FROM price_history
+            WHERE collection_id = %s
+            ORDER BY crawled_at DESC LIMIT %s OFFSET %s
+        """, (collection_id, page_size, offset))
+        items = cursor.fetchall()
+    
+    return ApiResponse.success(
+        PaginatedData.create(items, page, page_size, total)
+    )

api/routes/health.py

@@ -0,0 +1,37 @@
+"""健康检查路由"""
+from fastapi import APIRouter
+from datetime import datetime
+from database import db
+
+from api.middleware.response import ApiResponse
+
+router = APIRouter(tags=["health"])
+
+@router.get("/health")
+async def health_check():
+    """服务健康检查"""
+    health_status = {
+        "status": "healthy",
+        "timestamp": datetime.now().isoformat() + "Z",
+        "services": {}
+    }
+    
+    # 检查数据库
+    try:
+        with db.get_cursor() as cursor:
+            cursor.execute("SELECT 1")
+            health_status["services"]["database"] = "connected"
+    except Exception as e:
+        health_status["services"]["database"] = f"error: {str(e)}"
+        health_status["status"] = "degraded"
+    
+    return ApiResponse.success(health_status)
+
+@router.get("/")
+async def root():
+    """API 根路径"""
+    return ApiResponse.success({
+        "name": "CoolBotDataSys API",
+        "version": "1.0.0",
+        "docs": "/docs"
+    })

api/routes/prices.py

@@ -0,0 +1,68 @@
+"""价格路由"""
+from fastapi import APIRouter, Query
+from typing import Optional
+from datetime import datetime
+
+from api.models.response import PriceHistoryResponse
+from api.middleware.response import ApiResponse, PaginatedData
+from database import db
+
+router = APIRouter(prefix="/api/v1/prices", tags=["prices"])
+
+@router.get("/latest", response_model=ApiResponse)
+async def get_latest_prices(
+    category: Optional[str] = Query(None, description="品类筛选"),
+    source: Optional[str] = Query(None, description="来源筛选"),
+    limit: int = Query(50, ge=1, le=200, description="返回数量")
+):
+    """获取最新价格记录"""
+    with db.get_cursor() as cursor:
+        sql = """
+            SELECT ph.*, c.name as collection_name, c.category, c.serial_number
+            FROM price_history ph
+            LEFT JOIN collections c ON ph.collection_id = c.id
+        """
+        params = []
+        where_clauses = []
+        
+        if category:
+            where_clauses.append("c.category = %s")
+            params.append(category)
+        if source:
+            where_clauses.append("ph.source = %s")
+            params.append(source)
+        
+        if where_clauses:
+            sql += " WHERE " + " AND ".join(where_clauses)
+        
+        sql += " ORDER BY ph.crawled_at DESC LIMIT %s"
+        params.append(limit)
+        
+        cursor.execute(sql, params)
+        items = cursor.fetchall()
+    
+    return ApiResponse.success(items)
+
+@router.get("/trend", response_model=ApiResponse)
+async def get_price_trend(
+    collection_id: int = Query(..., description="藏品ID"),
+    days: int = Query(30, ge=1, le=365, description="统计天数")
+):
+    """获取价格趋势（按天统计）"""
+    with db.get_cursor() as cursor:
+        cursor.execute("""
+            SELECT 
+                DATE(ph.crawled_at) as date,
+                AVG(ph.price) as avg_price,
+                MIN(ph.price) as min_price,
+                MAX(ph.price) as max_price,
+                COUNT(*) as record_count
+            FROM price_history ph
+            WHERE ph.collection_id = %s
+              AND ph.crawled_at >= DATE_SUB(NOW(), INTERVAL %s DAY)
+            GROUP BY DATE(ph.crawled_at)
+            ORDER BY date ASC
+        """, (collection_id, days))
+        items = cursor.fetchall()
+    
+    return ApiResponse.success(items)

config.py

@@ -0,0 +1,45 @@
+"""配置加载模块"""
+import os
+import yaml
+from pathlib import Path
+
+class Config:
+    _instance = None
+    _config = None
+    
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._load_config()
+        return cls._instance
+    
+    def _load_config(self):
+        config_path = Path(__file__).parent / "config" / "config.yaml"
+        with open(config_path, "r", encoding="utf-8") as f:
+            self._config = yaml.safe_load(f)
+    
+    @property
+    def database(self):
+        return self._config.get("database", {})
+    
+    @property
+    def redis(self):
+        return self._config.get("redis", {})
+    
+    @property
+    def app(self):
+        return self._config.get("app", {})
+    
+    @property
+    def crawlers(self):
+        return self._config.get("crawlers", {})
+    
+    @property
+    def notification(self):
+        return self._config.get("notification", {})
+    
+    @property
+    def tasks(self):
+        return self._config.get("tasks", {})
+
+config = Config()

config/config.yaml

@@ -0,0 +1,44 @@
+# CoolBotDataSys 配置文件
+
+database:
+  host: "127.0.0.1"
+  port: 3306
+  user: "root"
+  password: "Coolbot123"
+  database: "coolbot_data"
+  charset: "utf8mb4"
+
+redis:
+  host: "127.0.0.1"
+  port: 6379
+  db: 0
+
+app:
+  host: "0.0.0.0"
+  port: 8080
+  debug: true
+  log_level: "INFO"
+
+crawlers:
+  yichens:
+    name: "一尘网"
+    base_url: "https://www.yichens.com"
+    enabled: true
+    interval_minutes: 60
+  aicang:
+    name: "爱藏网"
+    base_url: "https://www.aicang.com"
+    enabled: false
+    interval_minutes: 60
+
+notification:
+  feishu:
+    enabled: true
+    webhook_url: ""
+
+tasks:
+  daily_report:
+    enabled: true
+    cron: "0 9 * * *"
+  crawl_interval:
+    cron: "*/30 * * * *"

crawlers/__init__.py

@@ -0,0 +1 @@
+"""爬虫模块"""

crawlers/yichens_spider.py

@@ -0,0 +1,140 @@
+"""一尘网爬虫 - 龙钞价格数据采集"""
+import requests
+from bs4 import BeautifulSoup
+import re
+import time
+import logging
+from datetime import datetime
+from database import db
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+class YichensSpider:
+    """一尘网爬虫"""
+    
+    def __init__(self):
+        self.name = "一尘网"
+        self.source = "yichens"
+        self.base_url = "https://www.yichens.com"
+        self.headers = {
+            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
+        }
+    
+    def parse_price(self, price_str: str) -> float:
+        """解析价格字符串"""
+        if not price_str:
+            return 0.0
+        match = re.search(r"[\d.]+", price_str.replace(",", ""))
+        return float(match.group()) if match else 0.0
+    
+    def crawl_longchao_prices(self) -> list:
+        """
+        采集龙钞价格数据
+        
+        Returns:
+            list: 采集到的价格数据列表
+        """
+        logger.info("开始采集龙钞价格数据...")
+        items = []
+        
+        try:
+            # TODO: 根据实际网站结构调整URL和解析逻辑
+            url = f"{self.base_url}/nbbs/list?category=longchao"
+            logger.info(f"请求URL: {url}")
+            
+            response = requests.get(url, headers=self.headers, timeout=10)
+            
+            if response.status_code == 200:
+                soup = BeautifulSoup(response.text, "lxml")
+                logger.info(f"页面获取成功，内容长度: {len(response.text)}")
+                # TODO: 根据实际网页结构解析价格数据
+            else:
+                logger.warning(f"HTTP状态码: {response.status_code}")
+                
+        except requests.RequestException as e:
+            logger.error(f"网络请求失败: {e}")
+        
+        return items
+    
+    def save_to_db(self, items: list) -> int:
+        """保存到数据库"""
+        if not items:
+            logger.info("无新数据需要保存")
+            return 0
+        
+        saved = 0
+        with db.get_cursor() as cursor:
+            for item in items:
+                try:
+                    cursor.execute("""
+                        INSERT INTO collections (name, category, serial_number, cost_price, status)
+                        VALUES (%s, %s, %s, %s, %s)
+                        ON DUPLICATE KEY UPDATE cost_price = VALUES(cost_price), updated_at = NOW()
+                    """, (
+                        item.get("name"),
+                        item.get("category"),
+                        item.get("serial_number"),
+                        item.get("price"),
+                        "in_collection"
+                    ))
+                    
+                    collection_id = cursor.lastrowid if cursor.lastrowid else 0
+                    
+                    cursor.execute("""
+                        INSERT INTO price_history (collection_id, source, price, price_unit, price_type, url)
+                        VALUES (%s, %s, %s, %s, %s, %s)
+                    """, (
+                        collection_id,
+                        self.source,
+                        item.get("price"),
+                        item.get("unit", "元/张"),
+                        item.get("price_type", "挂牌价"),
+                        item.get("url", "")
+                    ))
+                    
+                    saved += 1
+                except Exception as e:
+                    logger.error(f"保存失败: {e}")
+        
+        logger.info(f"成功保存 {saved} 条记录")
+        return saved
+    
+    def run(self) -> list:
+        """执行爬虫"""
+        log_id = self._log_start()
+        
+        try:
+            items = self.crawl_longchao_prices()
+            saved = self.save_to_db(items)
+            self._log_finish(log_id, "success", saved)
+            return items
+        except Exception as e:
+            logger.error(f"爬虫执行失败: {e}")
+            self._log_finish(log_id, "failed", 0, str(e))
+            return []
+    
+    def _log_start(self) -> int:
+        """记录爬虫开始"""
+        with db.get_cursor() as cursor:
+            cursor.execute("""
+                INSERT INTO crawl_logs (source, status, started_at)
+                VALUES (%s, %s, NOW())
+            """, (self.source, "running"))
+            return cursor.lastrowid
+    
+    def _log_finish(self, log_id: int, status: str, items_count: int, error: str = ""):
+        """记录爬虫结束"""
+        with db.get_cursor() as cursor:
+            cursor.execute("""
+                UPDATE crawl_logs 
+                SET status = %s, items_count = %s, error_message = %s, finished_at = NOW()
+                WHERE id = %s
+            """, (status, items_count, error, log_id))
+
+
+if __name__ == "__main__":
+    spider = YichensSpider()
+    spider.run()

database.py

@@ -0,0 +1,57 @@
+"""MySQL 数据库连接池模块"""
+import mysql.connector
+from mysql.connector import pooling
+from contextlib import contextmanager
+
+class Database:
+    _pool = None
+    
+    @classmethod
+    def get_pool(cls):
+        if cls._pool is None:
+            cls._pool = pooling.MySQLConnectionPool(
+                pool_name="coolbot_pool",
+                pool_size=5,
+                host="127.0.0.1",
+                port=3306,
+                user="root",
+                password="Coolbot123",
+                database="coolbot_data",
+                charset="utf8mb4"
+            )
+        return cls._pool
+    
+    @contextmanager
+    def get_connection(self):
+        conn = self.get_pool().get_connection()
+        try:
+            yield conn
+        finally:
+            conn.close()
+    
+    @contextmanager
+    def get_cursor(self, dictionary=True):
+        with self.get_connection() as conn:
+            cursor = conn.cursor(dictionary=dictionary)
+            try:
+                yield cursor
+                conn.commit()
+            except Exception as e:
+                conn.rollback()
+                raise e
+            finally:
+                cursor.close()
+
+db = Database()
+
+def test_connection():
+    """测试数据库连接"""
+    try:
+        with db.get_cursor() as cursor:
+            cursor.execute("SELECT 1 as test")
+            result = cursor.fetchone()
+            print(f"✅ 数据库连接成功: {result}")
+            return True
+    except Exception as e:
+        print(f"❌ 数据库连接失败: {e}")
+        return False

requirements.txt

@@ -0,0 +1,31 @@
+# CoolBotDataSys Python Dependencies
+
+# Web Framework
+fastapi>=0.109.0
+uvicorn[standard]>=0.27.0
+pydantic>=2.5.0
+
+# Database
+mysql-connector-python>=8.3.0
+redis>=5.0.0
+sqlalchemy>=2.0.0
+
+# Web Scraping
+scrapy>=2.11.0
+playwright>=1.40.0
+requests>=2.31.0
+beautifulsoup4>=4.12.0
+lxml>=5.1.0
+
+# Data Processing
+pandas>=2.1.0
+numpy>=1.26.0
+
+# Task Scheduling
+apscheduler>=3.10.0
+
+# Utilities
+python-dotenv>=1.0.0
+pyyaml>=6.0.0
+httpx>=0.26.0
+python-dateutil>=2.8.0