diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000..761a49f --- /dev/null +++ b/api/__init__.py @@ -0,0 +1,2 @@ +"""CoolBotDataSys API Module""" +__version__ = "1.0.0" diff --git a/api/main.py b/api/main.py new file mode 100644 index 0000000..313e571 --- /dev/null +++ b/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) diff --git a/api/middleware/response.py b/api/middleware/response.py new file mode 100644 index 0000000..8aa031c --- /dev/null +++ b/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 + } + ) diff --git a/api/models/__init__.py b/api/models/__init__.py new file mode 100644 index 0000000..ebb9f5d --- /dev/null +++ b/api/models/__init__.py @@ -0,0 +1,2 @@ +from .request import CollectionCreate, CollectionUpdate, PriceQuery +from .response import CollectionResponse, PriceHistoryResponse, CrawlLogResponse, StatisticsResponse diff --git a/api/models/request.py b/api/models/request.py new file mode 100644 index 0000000..956c66a --- /dev/null +++ b/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="每页数量") diff --git a/api/models/response.py b/api/models/response.py new file mode 100644 index 0000000..8ed7a94 --- /dev/null +++ b/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 = {} diff --git a/api/routes/__init__.py b/api/routes/__init__.py new file mode 100644 index 0000000..0d4e1b8 --- /dev/null +++ b/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 diff --git a/api/routes/collections.py b/api/routes/collections.py new file mode 100644 index 0000000..511daac --- /dev/null +++ b/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) + ) diff --git a/api/routes/health.py b/api/routes/health.py new file mode 100644 index 0000000..0fa1893 --- /dev/null +++ b/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" + }) diff --git a/api/routes/prices.py b/api/routes/prices.py new file mode 100644 index 0000000..f711616 --- /dev/null +++ b/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) diff --git a/config.py b/config.py new file mode 100644 index 0000000..3f54c2f --- /dev/null +++ b/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() diff --git a/config/config.yaml b/config/config.yaml new file mode 100644 index 0000000..eb0cadc --- /dev/null +++ b/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 * * * *" diff --git a/crawlers/__init__.py b/crawlers/__init__.py new file mode 100644 index 0000000..1962ccd --- /dev/null +++ b/crawlers/__init__.py @@ -0,0 +1 @@ +"""爬虫模块""" diff --git a/crawlers/yichens_spider.py b/crawlers/yichens_spider.py new file mode 100644 index 0000000..91463ee --- /dev/null +++ b/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() diff --git a/database.py b/database.py new file mode 100644 index 0000000..bac70d9 --- /dev/null +++ b/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 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6e7f54c --- /dev/null +++ b/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