fix: 修复yichens_spider_v4.py两处bug

1. post_time加入ON CONFLICT UPDATE SET - 未来重爬时可更新post_time 2. author_username提取从user_link改为<b>标签 - 修复用户名抓不到的问题
feat: 添加一尘网爬虫v4 - BeautifulSoup解析 + 日期格式修复
2026-04-05 10:13:13 +08:00 · 2026-04-05 09:45:30 +08:00
21 changed files with 798 additions and 1667 deletions
--- a/.env.example
+++ b/.env.example
@ -1,22 +0,0 @@
 # CoolBotDataSys 环境变量配置示例
 # 复制为 .env 后填入实际值
 # 数据库
 DB_HOST=pgm-bp1t1008h019ez6c.pg.rds.aliyuncs.com
 DB_PORT=5432
 DB_USER=coolbot
 DB_PASSWORD=your_password_here
 DB_NAME=coolbot_data
 # Redis
 REDIS_HOST=127.0.0.1
 REDIS_PORT=6379
 REDIS_DB=0
 # LLM (阿里云百炼)
 LLM_API_KEY=your_api_key_here
 LLM_API_URL=https://coding.dashscope.aliyuncs.com/v1/chat/completions
 LLM_MODEL=qwen3.5-plus
 # CORS 白名单 (逗号分隔)
 CORS_ORIGINS=http://47.98.171.101,http://47.98.171.101:80
--- a/README.md
+++ b/README.md
@ -1,119 +1,3 @@
-# CoolBotDataSys - 酷博特数据分析系统
+# CoolBotDataSys
-**版本：** v0.0.1  
+酷博特数据分析系统 - 多源数据采集、存储、清洗、分析全链路自动化平台
 **日期：** 2026-04-06
 多源数据采集、存储、清洗、分析全链路自动化平台。
 ## 功能模块
 | 模块 | 描述 |
 |------|------|
 | 爬虫引擎 | 一尘网连体纪念钞数据采集（每小时增量 + 每日全量） |
 | REST API | 藏品管理、价格追踪、帖子查询、统计报表 |
 | 数据分析 | 龙钞/马钞价格分类统计、求购/出售趋势分析 |
 | LLM 辅助 | 阿里云通义千问价格提取（可选） |
 ## 技术栈
 - **后端：** Python 3.12 + FastAPI + uvicorn
 - **数据库：** PostgreSQL（RDS）+ Redis（本地缓存）
 - **爬虫：** requests + BeautifulSoup（轻量定向采集）
 - **定时任务：** APScheduler + cron
 ## 目录结构
 ```
 coolbot-data/
 ├── api/                    # FastAPI 应用
 │   ├── main.py             # 应用入口、路由注册
 │   ├── middleware/         # 响应封装
 │   ├── models/             # Pydantic 模型
 │   └── routes/             # API 路由
 │       ├── collections.py  # 藏品管理
 │       ├── prices.py       # 价格查询
 │       ├── health.py       # 健康检查
 │       └── yichens.py      # 一尘数据（含统计报表）
 ├── crawlers/               # 爬虫模块
 │   ├── base.py             # 爬虫基类（分页/重试/延时）
 │   ├── crawl_today.py      # 今日增量采集
 │   └── yichens_spider.py   # 全量历史采集
 ├── scripts/                # 运维脚本
 │   ├── daily_crawl.sh      # 每日爬虫 cron
 │   ├── daily_report.py     # 飞书日报推送
 │   └── crawl_cron.sh       # cron 调度脚本
 ├── alembic/                # 数据库迁移
 ├── config/                 # YAML 配置
 ├── cache.py                # Redis 缓存
 ├── database.py             # PostgreSQL 连接
 ├── scheduler.py            # APScheduler 调度
 ├── llm_price_extract.py    # LLM 价格提取
 └── requirements.txt        # Python 依赖
 ```
 ## 快速启动
 ```bash
 # 1. 安装依赖
 pip install -r requirements.txt
 # 2. 配置环境变量
 cp .env.example .env
 # 编辑 .env 填入数据库密码等
 # 3. 启动 API 服务
 python -m uvicorn api.main:app --host 0.0.0.0 --port 8080
 # 4. 启动定时爬虫
 python scheduler.py
 ```
 ## API 文档
 启动服务后访问：http://localhost:8080/docs
 ### 核心接口
 | 接口 | 方法 | 描述 |
 |------|------|------|
 | /health | GET | 服务健康检查 |
 | /api/v1/statistics | GET | 系统统计 |
 | /api/v1/yichens/posts | GET | 帖子列表（支持分页/筛选） |
 | /api/v1/yichens/statistics | GET | 一尘数据统计 |
 | /api/v1/yichens/report/stats | GET | 龙钞分类统计（收购/出售细分） |
 | /api/v1/yichens/report/categories | GET | 分类统计（今/近1小时） |
 | /api/v1/collections | GET/POST | 藏品管理 |
 | /api/v1/prices/latest | GET | 最新价格 |
 | /api/v1/crawl-jobs/trigger | POST | 手动触发爬虫 |
 ## 数据库表
 - collections - 藏品主表
 - price_history - 价格历史
 - yichens_posts - 一尘帖子（核心）
 - yichens_users - 一尘用户
 - yichens_replies - 帖子回复
 - yichens_follows - 关注关系
 - crawl_logs - 爬虫运行日志
 - scheduled_tasks - 调度任务
 - sys_config - 系统配置
 ## 爬虫调度
 ```cron
 # 每15分钟增量采集
 */15 * * * * cd /root/coolbot-data && source venv/bin/activate && python3 crawlers/crawl_today.py >> logs/crawl_update.log 2>&1
 # 每天8点全量采集
 0 8 * * * cd /root/coolbot-data && source venv/bin/activate && python3 crawlers/yichens_spider.py >> logs/crawl_daily.log 2>&1
 ```
 ## 环境变量
 详见 .env.example
 ## Git
 ```bash
 git clone http://caibotmi:Caibotmi123@101.37.160.219/caibotmi/CoolBotDataSys.git
 ```
--- a/alembic.ini
+++ b/alembic.ini
@ -1,41 +0,0 @@
 [alembic]
 script_location = alembic
 prepend_sys_path = .
 version_path_separator = os
 sqlalchemy.url = postgresql://coolbot:Coolbot123@pgm-bp1t1008h019ez6c.pg.rds.aliyuncs.com:5432/coolbot_data
 [post_write_hooks]
 [loggers]
 keys = root,sqlalchemy,alembic
 [handlers]
 keys = console
 [formatters]
 keys = generic
 [logger_root]
 level = WARN
 handlers = console
 qualname =
 [logger_sqlalchemy]
 level = WARN
 handlers =
 qualname = sqlalchemy.engine
 [logger_alembic]
 level = INFO
 handlers =
 qualname = alembic
 [handler_console]
 class = StreamHandler
 args = (sys.stderr,)
 level = NOTSET
 formatter = generic
 [formatter_generic]
 format = %(levelname)-5.5s [%(name)s] %(message)s
 datefmt = %H:%M:%S
--- a/alembic/README
+++ b/alembic/README
@ -1 +0,0 @@
 Generic single-database configuration.
--- a/alembic/env.py
+++ b/alembic/env.py
@ -1,152 +0,0 @@
 from logging.config import fileConfig
 from sqlalchemy import engine_from_config, MetaData, Table, Column, String, Integer, BigInteger, Boolean, Text, Float, DateTime, Index
 from sqlalchemy import pool
 from alembic import context
 import sys
 import os
 sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
 config = context.config
 # 数据库配置
 db_config = {
    'host': 'pgm-bp1t1008h019ez6c.pg.rds.aliyuncs.com',
    'port': 5432,
    'user': 'coolbot',
    'password': 'Coolbot123',
    'database': 'coolbot_data'
 }
 config.set_main_option('sqlalchemy.url', 
    f"postgresql://{db_config['user']}:{db_config['password']}@"
    f"{db_config['host']}:{db_config['port']}/{db_config['database']}")
 if config.config_file_name is not None:
    fileConfig(config.config_file_name)
 # 定义 metadata（用于 autogenerate）
 metadata = MetaData()
 # 定义现有表结构
 Table('collections', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(255)),
    Column('category', String(100)),
    Column('series', String(100)),
    Column('issue_year', Integer),
    Column('description', Text),
    Column('image_url', String(500)),
    Column('created_at', DateTime),
    Column('updated_at', DateTime),
 )
 Table('yichens_posts', metadata,
    Column('id', BigInteger, primary_key=True, autoincrement=True),
    Column('post_id', String(100), unique=True),
    Column('topic_id', String(100)),
    Column('title', String(500)),
    Column('content', Text),
    Column('author_id', String(100)),
    Column('author_username', String(100)),
    Column('category', String(50)),
    Column('sub_category', String(50)),
    Column('post_type', String(20)),
    Column('price', Float),
    Column('price_unit', String(20)),
    Column('contact', String(200)),
    Column('special_types', String(200)),
    Column('has_lifebuoy', Boolean),
    Column('reply_count', Integer),
    Column('view_count', Integer),
    Column('post_url', String(500)),
    Column('created_at', DateTime),
    Column('updated_at', DateTime),
    Column('crawled_at', DateTime),
    Column('url', String(500)),
    Index('idx_yichens_posts_post_id', 'post_id'),
    Index('idx_yichens_posts_category', 'category'),
    Index('idx_yichens_posts_post_time', 'created_at'),
 )
 Table('price_history', metadata,
    Column('id', BigInteger, primary_key=True, autoincrement=True),
    Column('collection_id', Integer),
    Column('price', Float),
    Column('price_unit', String(20)),
    Column('source', String(50)),
    Column('url', String(500)),
    Column('post_type', String(20)),
    Column('special_types', String(200)),
    Column('author', String(100)),
    Column('contact', String(200)),
    Column('content', Text),
    Column('crawled_at', DateTime),
    Column('created_at', DateTime),
 )
 Table('crawl_logs', metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('source', String(50)),
    Column('status', String(20)),
    Column('items_count', Integer),
    Column('error_message', Text),
    Column('started_at', DateTime),
    Column('finished_at', DateTime),
 )
 Table('yichens_members', metadata,
    Column('user_id', String(100), primary_key=True),
    Column('username', String(100), unique=True),
    Column('transaction_level', String(50)),
    Column('credit_score', Integer),
    Column('rating_count', Integer),
    Column('post_count', Integer),
    Column('post_points', Integer),
    Column('has_business_license', Boolean),
    Column('license_info', Text),
    Column('identity_verified', Boolean),
    Column('real_name', String(100)),
    Column('verification_notes', Text),
    Column('phone', String(100)),
    Column('address', String(500)),
    Column('bank_accounts', Text),
    Column('alipay', String(200)),
    Column('registration_date', DateTime),
    Column('member_since', String(100)),
    Column('is_verified', Boolean),
    Column('status', String(20)),
    Column('created_at', DateTime),
 )
 target_metadata = metadata
 def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()
 def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection, 
            target_metadata=target_metadata
        )
        with context.begin_transaction():
            context.run_migrations()
 if context.is_offline_mode():
    run_migrations_offline()
 else:
    run_migrations_online()
--- a/alembic/script.py.mako
+++ b/alembic/script.py.mako
@ -1,28 +0,0 @@
 """${message}
 Revision ID: ${up_revision}
 Revises: ${down_revision | comma,n}
 Create Date: ${create_date}
 """
 from typing import Sequence, Union
 from alembic import op
 import sqlalchemy as sa
 ${imports if imports else ""}
 # revision identifiers, used by Alembic.
 revision: str = ${repr(up_revision)}
 down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
 branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
 depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
 def upgrade() -> None:
    """Upgrade schema."""
    ${upgrades if upgrades else "pass"}
 def downgrade() -> None:
    """Downgrade schema."""
    ${downgrades if downgrades else "pass"}
--- a/alembic/versions/9c08601089e0_baseline_v1.py
+++ b/alembic/versions/9c08601089e0_baseline_v1.py
@ -1,28 +0,0 @@
 """baseline_v1
 Revision ID: 9c08601089e0
 Revises: fc320cec7fe7
 Create Date: 2026-04-05 10:38:55.510964
 """
 from typing import Sequence, Union
 from alembic import op
 import sqlalchemy as sa
 # revision identifiers, used by Alembic.
 revision: str = '9c08601089e0'
 down_revision: Union[str, Sequence[str], None] = 'fc320cec7fe7'
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 def upgrade() -> None:
    """Upgrade schema."""
    pass
 def downgrade() -> None:
    """Downgrade schema."""
    pass
--- a/api/main.py
+++ b/api/main.py
@ -106,7 +106,7 @@ async def get_statistics():
        # 今日新增价格记录
        cursor.execute("""
            SELECT COUNT(*) as cnt FROM price_history 
-            WHERE DATE(crawled_at) = CURRENT_DATE
+            WHERE DATE(crawled_at) = CURDATE()
        """)
        today_price_records = cursor.fetchone()["cnt"]
--- a/api/middleware/response.py
+++ b/api/middleware/response.py
@ -31,10 +31,6 @@ class ApiResponse(BaseModel, Generic[T]):
 class PaginatedData(BaseModel, Generic[T]):
    """分页数据封装"""
    items: list[T]
    total: int
    page: int
    page_size: int
    total_pages: int
    pagination: dict
    @classmethod
@ -42,10 +38,6 @@ class PaginatedData(BaseModel, Generic[T]):
        total_pages = (total + page_size - 1) // page_size if page_size > 0 else 0
        return cls(
            items=items,
            total=total,
            page=page,
            page_size=page_size,
            total_pages=total_pages,
            pagination={
                "page": page,
                "page_size": page_size,
--- a/api/routes/yichens.py
+++ b/api/routes/yichens.py
@ -4,7 +4,7 @@ from typing import Optional, List
 from datetime import datetime
 from api.middleware.response import ApiResponse, PaginatedData
-from database import db, get_db
+from database import db
 router = APIRouter(prefix="/api/v1/yichens", tags=["yichens"])
@ -15,7 +15,7 @@ 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)
+    page_size: int = Query(20, ge=1, le=100)
 ):
    """获取一尘用户列表"""
    offset = (page - 1) * page_size
@ -33,7 +33,7 @@ async def list_yichens_users(
    with db.get_cursor() as cursor:
        cursor.execute(f"SELECT COUNT(*) as total FROM yichens_users{where_sql}", params)
-        total = cursor.fetchone()[0]
+        total = cursor.fetchone()["total"]
        cursor.execute(f"""
            SELECT * FROM yichens_users{where_sql}
@ -110,58 +110,47 @@ async def list_yichens_posts(
    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)
+    page_size: int = Query(20, ge=1, le=100)
 ):
    """获取一尘帖子列表"""
    offset = (page - 1) * page_size
-    where_clauses = ["1=1"]
+    where_clauses = []
    params = []
    if category:
        where_clauses.append("category = %s")
        params.append(category)
-    if post_type == 'other':
+    if post_type:
        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)")
+        where_clauses.append("MATCH(title, content) AGAINST(%s IN NATURAL LANGUAGE MODE)")
-        params.extend([f"%{keyword}%", f"%{keyword}%"])
+        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)
    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)
+    where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
-    with get_db() as conn:
+    with db.get_cursor() as cursor:
-        cur = conn.cursor()
+        cursor.execute(f"SELECT COUNT(*) as total FROM yichens_posts{where_sql}", params)
-        cur.execute(f"SELECT COUNT(*) as total FROM yichens_posts{where_sql}", params)
+        total = cursor.fetchone()["total"]
-        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])
+        cursor.execute(f"""
-        rows = cur.fetchall()
+            SELECT * FROM yichens_posts{where_sql}
-        cols = [desc[0] for desc in cur.description]
+            ORDER BY created_at DESC LIMIT %s OFFSET %s
-        items = [dict(zip(cols, row)) for row in rows]
+        """, params + [page_size, offset])
-        cur.close()
+        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):
    """获取一尘帖子详情"""
@ -247,7 +236,7 @@ async def get_post_replies(
            "SELECT COUNT(*) as total FROM yichens_replies WHERE post_id = %s",
            (post_id,)
        )
-        total = cursor.fetchone()[0]
+        total = cursor.fetchone()["total"]
        cursor.execute("""
            SELECT * FROM yichens_replies
@ -320,206 +309,16 @@ async def get_yichens_statistics():
        cursor.execute("""
            SELECT DATE(crawled_at) as date, COUNT(*) as count
            FROM yichens_posts
-            WHERE crawled_at >= NOW() - INTERVAL '7 days'
+            WHERE crawled_at >= DATE_SUB(NOW(), INTERVAL 7 DAY)
            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})
--- a/cache.py
+++ b/cache.py
@ -1,134 +0,0 @@
 """
 Redis 缓存模块 - 修复版
 实现爬虫结果缓存、API 响应缓存、去重等功能
 """
 import redis
 import json
 import logging
 from typing import Optional, Any
 import os
 import sys
 sys.path.insert(0, os.path.dirname(__file__))
 logger = logging.getLogger(__name__)
 # 使用 Config 类
 from config import config
 redis_config = config.redis
 class RedisCache:
    _instance = None
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._init_client()
        return cls._instance
    def _init_client(self):
        try:
            self.client = redis.Redis(
                host=redis_config.get('host', '127.0.0.1'),
                port=redis_config.get('port', 6379),
                db=redis_config.get('db', 0),
                decode_responses=True,
                socket_timeout=5,
                socket_connect_timeout=5,
                retry_on_timeout=True
            )
            self.client.ping()
            logger.info("Redis connected successfully")
        except redis.ConnectionError as e:
            logger.warning(f"Redis connection failed: {e}, caching disabled")
            self.client = None
    @property
    def is_available(self) -> bool:
        if not self.client:
            return False
        try:
            self.client.ping()
            return True
        except:
            return False
    def get(self, key: str) -> Optional[Any]:
        if not self.is_available:
            return None
        try:
            val = self.client.get(key)
            if val:
                return json.loads(val)
            return None
        except Exception as e:
            logger.warning(f"Cache get error: {e}")
            return None
    def set(self, key: str, value: Any, ttl: int = 300):
        if not self.is_available:
            return False
        try:
            self.client.setex(key, ttl, json.dumps(value, default=str))
            return True
        except Exception as e:
            logger.warning(f"Cache set error: {e}")
            return False
    def delete(self, key: str):
        if not self.is_available:
            return False
        try:
            self.client.delete(key)
            return True
        except Exception as e:
            logger.warning(f"Cache delete error: {e}")
            return False
    def invalidate_pattern(self, pattern: str):
        if not self.is_available:
            return 0
        count = 0
        try:
            for key in self.client.scan_iter(match=pattern):
                self.client.delete(key)
                count += 1
            return count
        except Exception as e:
            logger.warning(f"Cache invalidate error: {e}")
            return count
    def get_post_cached(self, post_id: str) -> Optional[dict]:
        return self.get(f"post:{post_id}")
    def set_post_cached(self, post_id: str, data: dict, ttl: int = 300):
        return self.set(f"post:{post_id}", data, ttl)
    def is_post_crawled(self, post_id: str) -> bool:
        if not self.is_available:
            return False
        try:
            return self.client.exists(f"crawled:{post_id}") > 0
        except:
            return False
    def mark_post_crawled(self, post_id: str, ttl: int = 86400):
        if not self.is_available:
            return False
        try:
            self.client.setex(f"crawled:{post_id}", ttl, "1")
            return True
        except:
            return False
    def get_statistics_cached(self) -> Optional[dict]:
        return self.get("yichens:statistics")
    def set_statistics_cached(self, data: dict, ttl: int = 60):
        return self.set("yichens:statistics", data, ttl)
    def invalidate_statistics(self):
        return self.delete("yichens:statistics")
 cache = RedisCache()
--- a/crawlers/backfill_special.py
+++ b/crawlers/backfill_special.py
@ -1,68 +0,0 @@
 """补充更新 special_types 和 number_features（基于标题重算）"""
 import re
 import yaml
 from datetime import datetime
 def load_config():
    with open('/root/coolbot-data/config.yaml', 'r') as f:
        return yaml.safe_load(f)
 config = load_config()
 DB_CONFIG = config['database']
 def get_db_conn():
    import psycopg2
    return psycopg2.connect(
        host=DB_CONFIG['host'], port=DB_CONFIG.get('port', 5432),
        user=DB_CONFIG['user'], password=DB_CONFIG['password'],
        database=DB_CONFIG['database']
    )
 FEATURE_MAP = {
    '倒置': ['倒置'], '如意': ['如意'], '朦胧': ['朦胧'], '金马': ['金马'], '金山': ['金山'],
    '天马': ['天马'], '钻石': ['钻石'], '永恒': ['永恒'], '无4': ['无4', '无四'],
    '无47': ['无47'], '无247': ['无247'], '无34': ['无34'], '无347': ['无347'],
    '带4': ['带4'], '豹子': ['豹子'], '狮子': ['狮子'], '老虎': ['老虎'],
    '大象': ['大象'], '生日': ['生日'], '满号': ['满号'], '首日': ['首日'],
 }
 SPECIAL_MAP = {
    '标十': ['标十'], '标百': ['标百'],
    '刀': ['刀', '刀货'], '单张': ['单张', '散张'],
    '捆': ['捆'], '千连': ['千连'], '百连': ['百连'],
    '救生圈': ['大救生圈', '救生圈'],
 }
 def calc_special(title):
    if not title: return None
    found = [k for k, words in SPECIAL_MAP.items() if any(w in title for w in words)]
    return '|'.join(found) if found else None
 def calc_features(title):
    if not title: return None
    found = [k for k, words in FEATURE_MAP.items() if any(w in title for w in words)]
    return '|'.join(found) if found else None
 conn = get_db_conn()
 cur = conn.cursor()
 cur.execute("SELECT id, title FROM yichens_posts WHERE special_types IS NULL OR number_features IS NULL")
 rows = cur.fetchall()
 print(f'需要更新: {len(rows)} 条')
 updated = 0
 for rid, title in rows:
    sp = calc_special(title)
    ft = calc_features(title)
    cur.execute("""
        UPDATE yichens_posts SET special_types = COALESCE(%s, special_types),
        number_features = COALESCE(%s, number_features)
        WHERE id = %s
    """, (sp, ft, rid))
    updated += 1
    if updated % 200 == 0:
        print(f'已更新 {updated}/{len(rows)}')
 conn.commit()
 cur.close()
 conn.close()
 print(f'完成! 共更新 {updated} 条')
--- a/crawlers/crawl_today.py
+++ b/crawlers/crawl_today.py
@ -1,401 +0,0 @@
 """只采集今天的新帖子 - 优化版"""
 import re
 import sys
 import os
 sys.path.insert(0, '/root/coolbot-data')
 import requests
 from bs4 import BeautifulSoup
 from datetime import datetime
 from crawlers.base import PaginationSpider
 from database import get_db
 def extract_post_content(html_bytes):
    """从详情页原始HTML提取帖子正文（GBK编码）。
    定位：找到含 min-height:200px 的 div，向下到 <hr />（签名）为止。
    """
    if isinstance(html_bytes, str):
        page = html_bytes
    else:
        page = html_bytes.decode('gbk', errors='replace')
    idx = page.find('min-height:200px')
    if idx == -1:
        return None
    div_start = page.rfind('<div', 0, idx)
    if div_start == -1:
        return None
    hr_idx = page.find('<hr />', idx)
    discl_idx = page.find('\u514d\u8d23\u5371\u58f0\u660e', idx)
    candidates = [x for x in [hr_idx, discl_idx] if x != -1]
    end = min(candidates) if candidates else (hr_idx if hr_idx != -1 else len(page))
    chunk = page[div_start:end]
    text = re.sub(r'<div[^>]*>', '\n', chunk)
    text = re.sub(r'</div>', '', text)
    text = re.sub(r'<p[^>]*>', '\n', text)
    text = re.sub(r'</p>', '', text)
    text = re.sub(r'<br\s*/?>', '\n', text)
    text = re.sub(r'<[^>]+>', '', text)
    text = text.replace('&nbsp;', ' ').replace('&#160;', ' ')
    text = text.replace('&amp;', '&').replace('&lt;', '<').replace('&gt;', '>')
    lines = [l.strip() for l in text.split('\n') if l.strip()]
    text = '\n'.join(lines)
    text = re.sub(r'\n\d{4}[/\-]\d{1,2}[/\-]\d{1,2}\s+\d{1,2}:\d{2}:\d{2}\s*$', '', text)
    return text if text else None
 import logging
 logger = logging.getLogger(__name__)
 class YichensTodaySpider(PaginationSpider):
    def __init__(self):
        super().__init__("一尘网今日采集", "pm001")
        self.base_url = "http://www.pm001.net"
        self.board_id = "151"
        self.forum_url = f"{self.base_url}/index.asp?boardid={self.board_id}"
        self.max_pages = 2
        self.encoding = "gbk"
        self.min_delay = 1.5
        self.max_delay = 3.0
    def get(self, url, **kwargs):
        try:
            self._random_delay()
            headers = {
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) 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",
            }
            response = requests.get(url, timeout=30, headers=headers, **kwargs)
            try:
                response.encoding = self.encoding
            except:
                response.encoding = "gb2312"
            return response
        except Exception as e:
            print(f"请求失败: {e}")
            return None
    def extract_posts_with_dates(self, html):
        """从列表页提取所有帖子ID和"发表于"时间，返回 [(post_id, date_str), ...]
        使用BeautifulSoup解析listtitle div，比正则更准确
        """
        soup = BeautifulSoup(html, 'html.parser')
        posts = []
        listtitle_divs = soup.find_all('div', class_='listtitle')
        for div in listtitle_divs:
            link = div.find('a', href=re.compile(r'dispbbs\.asp\?boardID=151&ID=\d+'))
            if not link:
                continue
            href = link.get('href', '')
            id_match = re.search(r'\&ID=(\d+)', href)
            if not id_match:
                continue
            post_id = id_match.group(1)
            title_attr = link.get('title', '')
            # title格式: 《标题》\n作者：xxx\n发表于：2026/4/5 9:35:00
            date_match = re.search(r'发表于：(\d{4}/\d{1,2}/\d{1,2})', title_attr)
            if date_match:
                date_str = date_match.group(1)
                posts.append((post_id, date_str))
        return posts
    def parse_post_detail(self, html, url):
        """解析详情页，提取真实发帖时间"""
        soup = BeautifulSoup(html, "html.parser")
        page_text = soup.get_text()
        match = re.search(r"\&ID=(\d+)", url)
        post_id = match.group(1) if match else None
        title = None
        title_elem = soup.find("title")
        if title_elem:
            title_text = title_elem.get_text(strip=True)
            if "[" in title_text:
                title = title_text.split("[")[0].strip()
        author = None
        m = re.search(r">>>\s*([^\s<]+)", page_text)
        if m:
            author = m.group(1).strip()
        created_at = None
        m = re.search(r"(\d{4}/\d{1,2}/\d{1,2}\s+\d{1,2}:\d{2}:\d{2})", page_text)
        if m:
            created_at = m.group(1).replace("/", "-")
        phones = re.findall(r'1[3-9]\d{9,10}', page_text)
        contact = ",".join(dict.fromkeys(phones[:5])) if phones else None
        price = None
        price_unit = None
        if title:
            price, price_unit = self.extract_price(title)
        if not price:
            price, price_unit = self.extract_price(page_text)
        content = extract_post_content(html)
        special_types = []
        if title:
            if "救生圈" in title or "88" in title or "888" in title:
                special_types.append("救生圈")
            if "标十" in title:
                special_types.append("标十")
            if "标百" in title:
                special_types.append("标百")
        post_type = "normal"
        if title:
            if any(c in title for c in ["出", "售", "卖", "兑"]):
                post_type = "deal"
            elif any(c in title for c in ["求", "收", "购"]):
                post_type = "want"
        category = "其他"
        if title:
            if "龙钞" in title or "龙纪念" in title:
                category = "龙钞"
            elif "蛇钞" in title:
                category = "蛇钞"
            elif "马钞" in title:
                category = "马钞"
        # 从 postuserinfo div 的 <b> 标签提取用户名（跳过电话号码）
        if not author:
            userinfos = soup.find_all('div', class_='postuserinfo')
            for ui in userinfos:
                for b in ui.find_all('b'):
                    text = b.get_text(strip=True)
                    if text and not re.match(r'^1\d{10}$', text) and len(text) > 1:
                        author = text
                        break
                if author:
                    break
        return {
            "post_id": post_id,
            "title": title,
            "content": content,
            "author_username": author,
            "price": price,
            "price_unit": price_unit,
            "contact": contact,
            "post_time": created_at,
            "special_types": "|".join(special_types) if special_types else None,
            "category": category,
            "post_type": post_type,
            "url": f"http://www.pm001.net/dispbbs.asp?boardID={self.board_id}&ID={post_id}&page=1",
        }
    def extract_price(self, text):
        if not text:
            return None, None
        patterns = [
            r"(\d+)\s*元\s*(?:一张|一刀|一条|单张)?",
            r"[收售出求兑买]\s*(\d+)\s*元?",
        ]
        for pattern in patterns:
            m = re.search(pattern, text)
            if m:
                try:
                    val = float(m.group(1))
                    if val < 0 or val > 999999999:
                        continue
                    price_unit = "元"
                    if "张" in text:
                        price_unit = "元/张"
                    elif "刀" in text:
                        price_unit = "元/刀"
                    elif "条" in text:
                        price_unit = "元/条"
                    return val, price_unit
                except:
                    pass
        return None, None
    def crawl_today(self):
        """全量采集：爬取第1-max_pages页所有帖子，不过滤日期
        用于一次性补全所有帖子的post_time等字段
        """
        today = datetime.now().date()
        today_str = today.strftime("%Y/%m/%d").lstrip('0').replace('/0', '/')
        print(f"开始全量采集第1-{self.max_pages}页 (今天: {today_str})")
        saved = 0
        checked = 0
        for page in range(1, self.max_pages + 1):
            page_url = f"{self.forum_url}&page={page}"
            print(f"\n扫描第 {page} 页...")
            response = self.get(page_url)
            if not response:
                continue
            posts = self.extract_posts_with_dates(response.text)
            print(f"  该页共 {len(posts)} 条帖子")
            for post_id, date_str in posts:
                checked += 1
                # 标准化日期
                normalized_date = date_str.lstrip('0').replace('/0', '/')
                # 访问详情页
                detail_url = f"{self.base_url}/dispbbs.asp?boardID={self.board_id}&ID={post_id}&page=1"
                detail_response = self.get(detail_url)
                if not detail_response:
                    continue
                detail = self.parse_post_detail(detail_response.text, detail_url)
                # 验证详情页的真实发帖时间
                post_time_str = detail.get('post_time', '')
                if not post_time_str:
                    continue
                try:
                    post_time = datetime.strptime(post_time_str, '%Y-%m-%d %H:%M:%S').date()
                    is_today = (post_time == today)
                    label = f"今日{normalized_date}" if is_today else f"历史{post_time}"
                except:
                    label = "时间异常"
                # 保存（ON CONFLICT会更新post_time等字段）
                if self.save_full_post(detail):
                    saved += 1
                    print(f"  [保存] {post_id} - {detail.get('title', '')[:25]}... ({label}中已存{saved}条)")
                else:
                    print(f"  [失败] {post_id}")
            print(f"  页码 {page} 完成: 检查{len(posts)}条")
        print(f"\n===== 采集完成 =====")
        print(f"总计检查: {checked} 条")
        print(f"成功保存: {saved} 条")
        return saved
    def save_full_post(self, post):
        if not post or not post.get("post_id"):
            return False
        sql = """
            INSERT INTO yichens_posts (
                post_id, title, content, category, post_type,
                price, price_unit, special_types, number_features,
                author_username, author_id, contact,
                has_lifebuoy, reply_count, view_count,
                post_time, crawled_at, updated_at, url
            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s)
            ON CONFLICT (post_id) DO UPDATE SET
                title = EXCLUDED.title,
                content = EXCLUDED.content,
                category = EXCLUDED.category,
                post_type = EXCLUDED.post_type,
                price = EXCLUDED.price,
                price_unit = EXCLUDED.price_unit,
                special_types = EXCLUDED.special_types,
                author_username = EXCLUDED.author_username,
                contact = EXCLUDED.contact,
                has_lifebuoy = EXCLUDED.has_lifebuoy,
                post_time = EXCLUDED.post_time,
                updated_at = NOW(),
                crawled_at = NOW(),
                url = EXCLUDED.url
        """
        has_lifebuoy = post.get("special_types") and "救生圈" in post.get("special_types", "")
        try:
            with get_db() as conn:
                if not conn:
                    print("数据库连接失败")
                    return False
                cur = conn.cursor()
                cur.execute(sql, (
                    post.get("post_id"),
                    post.get("title"),
                    post.get("content"),
                    post.get("category"),
                    post.get("post_type"),
                    post.get("price"),
                    post.get("price_unit"),
                    post.get("special_types"),
                    post.get("number_features"),
                    post.get("author_username"),
                    f"user_{post.get('post_id')}",
                    post.get("contact"),
                    has_lifebuoy,
                    0,
                    0,
                    post.get("post_time"),
                    post.get("url"),
                ))
                conn.commit()
                cur.close()
                return True
        except Exception as e:
            print(f"保存失败: {e}")
            return False
    def run(self):
        log_id = self._log_start()
        try:
            result = self.crawl_today()
            self._log_finish(log_id, "success", result)
            return result
        except Exception as e:
            print(f"爬虫异常: {e}")
            self._log_finish(log_id, "failed", 0, str(e))
            return 0
    def _log_start(self):
        try:
            with get_db() as conn:
                cur = conn.cursor()
                cur.execute(
                    "INSERT INTO crawl_logs (spider_name, status, started_at) VALUES (%s, %s, NOW()) RETURNING id",
                    (self.source, "running")
                )
                log_id = cur.fetchone()[0]
                conn.commit()
                cur.close()
                return log_id
        except Exception as e:
            print(f"记录日志失败: {e}")
            return None
    def _log_finish(self, log_id, status, items_count, error=""):
        if log_id is None:
            return
        try:
            with get_db() as conn:
                cur = conn.cursor()
                cur.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)
                )
                conn.commit()
                cur.close()
        except Exception as e:
            print(f"更新日志失败: {e}")
 if __name__ == "__main__":
    spider = YichensTodaySpider()
    spider.run()
--- a/crawlers/yichens_spider.py
+++ b/crawlers/yichens_spider.py
@ -1,7 +1,5 @@
 """一尘网爬虫 - 适配 pm001.net 连体纪念钞板块"""
 import re
 import httpx
 import json
 from typing import Optional, Dict, List, Any
 from datetime import datetime, timedelta
 from bs4 import BeautifulSoup
@ -209,78 +207,20 @@ class YichensSpider(PaginationSpider):
            'created_at': created_at,
        }
    def _extract_price_with_llm(self, title: str) -> dict:
        """Use LLM to extract price from title"""
        prompt = f"""从以下钱币收藏帖子标题中提取信息：
 标题："{title}"
 提取：post_type(want求购/deal出售/空), price(价格数字或null), unit(元/张、元/刀、元/条、元/套、元等)
 回答JSON格式：{{"post_type":"want/deal/","price":数字,"unit":"元/张等"}}"""
        try:
            r = httpx.post(
                "https://coding.dashscope.aliyuncs.com/v1/chat/completions",
                headers={
                    "Authorization": "Bearer sk-sp-d5ce68bb203e48ca857c2aea25255b26",
                    "Content-Type": "application/json"
                },
                json={
                    "model": "qwen3.5-plus",
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.1
                },
                timeout=30.0
            )
            if r.status_code == 200:
                result = r.json()
                content_text = result['choices'][0]['message']['content'].strip()
                if '{' in content_text:
                    json_str = content_text[content_text.find('{'):content_text.rfind('}')+1]
                    return json.loads(json_str)
        except Exception as e:
            print(f"LLM error: {e}")
        return {}
    def _extract_price(self, text: str) -> Optional[float]:
        """从标题提取价格"""
-        # 第一优先级：明确的价格单位（元、万）
+        patterns = [
-        patterns_with_unit = [
+            r'(\d+)\s*[万Ww]?\s*元',
-            r'(\d+(?:\.\d+)?)\s*[万Ww]?\s*元',  # 123元, 1.5万元
+            r'(\d+)\s*-\s*(\d+)\s*[万Ww]?',
-            r'[每单共总]\s*(\d+(?:\.\d+)?)\s*元',  # 共123元
+            r'[收售出求兑买]\s*(\d+)',
            r'(\d+(?:\.\d+)?)\s*元\s*(?:每|每张|每刀|每条|每套)',  # 123元每张
        ]
-        for pattern in patterns_with_unit:
+        for pattern in patterns:
            m = re.search(pattern, text)
            if m:
                try:
-                    val = float(m.group(1))
+                    return float(m.group(1))
                    if val > 0:
                        return val
                except:
                    pass
        # 第二优先级：明确标价的（价格X元）
        price_keyword = r'(?:价格|价|售价|收价|成交价)[：:]\s*(\d+(?:\.\d+)?)'
        m = re.search(price_keyword, text)
        if m:
            try:
                val = float(m.group(1))
                if val > 0:
                    return val
            except:
                pass
        # 第三优先级：数字+万（万前面肯定是价格）
        wan_pattern = r'(\d+(?:\.\d+)?)\s*万'
        m = re.search(wan_pattern, text)
        if m:
            try:
                val = float(m.group(1))
                if val > 0:
                    return val
            except:
                pass
        return None
    def _extract_price_unit(self, text: str) -> Optional[str]:
@ -297,94 +237,6 @@ class YichensSpider(PaginationSpider):
            return '元/万'
        return '元'
    def batch_llm_price_update(self) -> int:
        """批量处理今日缺少价格的帖子，返回更新数量"""
        from database import get_db
        import httpx, json
        # 找出今日爬取的价格为空的帖子
        with get_db() as conn:
            cur = conn.cursor()
            cur.execute("""
                SELECT id, title FROM yichens_posts 
                WHERE post_time::date = CURRENT_DATE 
                AND (price IS NULL OR price = 0)
                AND title IS NOT NULL
                LIMIT 50
            """)
            posts = cur.fetchall()
            cur.close()
        if not posts:
            print(f"No posts needing LLM price extraction")
            return 0
        print(f"Found {len(posts)} posts needing LLM price extraction")
        # 批量提取（每批10条）
        batch_size = 10
        updated = 0
        for i in range(0, len(posts), batch_size):
            batch = posts[i:i+batch_size]
            # 构建批量prompt
            titles_text = '\n'.join([f'{p[0]}|{p[1]}' for p in batch])
            prompt = f"""分析以下钱币收藏帖子的标题，提取实际交易价格。
 格式要求：
 - 如果有明确价格（数字+元/张、元/刀、元/条、元/套等），提取价格数字
 - "求2组"不是价格，是数量
 - "765出一组"中765如果是价格则提取
 - 如果无法判断实际交易价格，填null
 标题列表：
 {titles_text}
 回答JSON数组格式（只回答JSON，不要其他内容）：
 [{{"id":帖子ID,"price":数字或null,"unit":"元/张等"}},...]"""
            try:
                r = httpx.post(
                    "https://coding.dashscope.aliyuncs.com/v1/chat/completions",
                    headers={
                        "Authorization": "Bearer sk-sp-d5ce68bb203e48ca857c2aea25255b26",
                        "Content-Type": "application/json"
                    },
                    json={
                        "model": "qwen3.5-plus",
                        "messages": [{"role": "user", "content": prompt}],
                        "temperature": 0.1
                    },
                    timeout=120.0
                )
                if r.status_code == 200:
                    result = r.json()
                    content_text = result['choices'][0]['message']['content'].strip()
                    if '[' in content_text:
                        json_str = content_text[content_text.find('['):content_text.rfind(']')+1]
                        results = json.loads(json_str)
                        with get_db() as conn:
                            cur = conn.cursor()
                            for res in results:
                                if res.get('price') and res['price'] > 0:
                                    cur.execute(
                                        "UPDATE yichens_posts SET price = %s, price_unit = %s WHERE id = %s",
                                        (float(res['price']), res.get('unit', '元'), res['id'])
                                    )
                                    updated += 1
                            cur.close()
                        print(f"Batch {i//batch_size + 1}: updated {len(results)} posts")
            except Exception as e:
                print(f"Batch {i//batch_size + 1} error: {e}")
        print(f"Total LLM price updates: {updated}")
        return updated
    def save_post(self, post: Dict) -> bool:
        """保存帖子到数据库"""
        if not post or not post.get('post_id'):
@ -490,8 +342,6 @@ class YichensSpider(PaginationSpider):
        try:
            posts = self.crawl_forum(max_pages, crawl_detail)
            print(f"Crawled {len(posts)} posts, running LLM price extraction...")
            self.batch_llm_price_update()
            self._log_finish(log_id, "success", len(posts))
            return posts
        except Exception as e:
--- a/crawlers/yichens_spider_v4.py
+++ b/crawlers/yichens_spider_v4.py
@ -0,0 +1,726 @@
 """一尘网爬虫 v4 - 完整版
 - 爬取帖子 + 所有楼层用户信息
 - 优化字段识别（category/special_types/number_features/price）
 - 增量去重 + 日志记录 + 防封延迟
 """
 import requests
 from bs4 import BeautifulSoup
 import re
 import time
 import yaml
 import random
 import json
 from datetime import datetime
 from concurrent.futures import ThreadPoolExecutor, as_completed
 def load_config():
    with open('/root/coolbot-data/config.yaml', 'r') as f:
        return yaml.safe_load(f)
 config = load_config()
 DB_CONFIG = config['database']
 BASE_URL = 'http://www4.pm001.net'
 BOARD_ID = '151'
 HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) 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 get_db_conn():
    import psycopg2
    return psycopg2.connect(
        host=DB_CONFIG['host'],
        port=DB_CONFIG.get('port', 5432),
        user=DB_CONFIG['user'],
        password=DB_CONFIG['password'],
        database=DB_CONFIG['database']
    )
 def get_existing_post_ids():
    try:
        conn = get_db_conn()
        cur = conn.cursor()
        cur.execute('SELECT post_id FROM yichens_posts')
        existing = set(row[0] for row in cur.fetchall())
        cur.close()
        conn.close()
        return existing
    except Exception as e:
        print(f'获取已有post_id失败: {e}')
        return set()
 def log_crawl(spider_name, status, items_count, error=''):
    try:
        conn = get_db_conn()
        cur = conn.cursor()
        cur.execute("""
            INSERT INTO crawl_logs (spider_name, status, items_count, error_message, started_at, finished_at)
            VALUES (%s, %s, %s, %s, %s, %s)
        """, (spider_name, status, items_count, error, datetime.now(), datetime.now()))
        conn.commit()
        cur.close()
        conn.close()
    except Exception as e:
        print(f'写入爬虫日志失败: {e}')
 def save_post(post):
    try:
        conn = get_db_conn()
        cur = conn.cursor()
        sql = """
            INSERT INTO yichens_posts (
                post_id, title, content, category, post_type, price, price_unit,
                special_types, number_features, author_username, author_id,
                contact, has_lifebuoy, reply_count, view_count,
                post_time, crawled_at, url
            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON CONFLICT (post_id) DO UPDATE SET
                title = EXCLUDED.title,
                content = EXCLUDED.content,
                author_username = EXCLUDED.author_username,
                contact = EXCLUDED.contact,
                reply_count = EXCLUDED.reply_count,
                view_count = EXCLUDED.view_count,
                price = EXCLUDED.price,
                post_time = EXCLUDED.post_time,
                updated_at = CURRENT_TIMESTAMP
        """
        cur.execute(sql, (
            post.get('post_id'),
            post.get('title'),
            post.get('content'),
            post.get('category'),
            post.get('post_type'),
            post.get('price'),
            post.get('price_unit', '元'),
            post.get('special_types'),
            post.get('number_features'),
            post.get('author_username'),
            post.get('author_id'),
            post.get('contact'),
            post.get('has_lifebuoy', False),
            post.get('reply_count', 0),
            post.get('view_count', 0),
            post.get('post_time'),
            datetime.now(),
            post.get('url')
        ))
        conn.commit()
        cur.close()
        conn.close()
        return True
    except Exception as e:
        print(f'  DB error: {e}')
        return False
 def save_member(member):
    """保存会员信息到 yichens_members 表"""
    if not member or not member.get('user_id'):
        return False
    try:
        conn = get_db_conn()
        cur = conn.cursor()
        sql = """
            INSERT INTO yichens_members (
                user_id, username, transaction_level, credit_score,
                rating_count, post_count, post_points,
                has_business_license, real_name,
                phone, address, bank_accounts, alipay,
                registration_date, updated_at
            ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
            ON CONFLICT (user_id) DO UPDATE SET
                username = EXCLUDED.username,
                transaction_level = EXCLUDED.transaction_level,
                credit_score = EXCLUDED.credit_score,
                rating_count = EXCLUDED.rating_count,
                post_count = EXCLUDED.post_count,
                has_business_license = EXCLUDED.has_business_license,
                phone = EXCLUDED.phone,
                updated_at = CURRENT_TIMESTAMP
        """
        bank_accounts_json = json.dumps(member.get('bank_accounts', []), ensure_ascii=False)
        cur.execute(sql, (
            member.get('user_id'),
            member.get('username'),
            member.get('transaction_level'),
            member.get('credit_score'),
            member.get('rating_count', 0),
            member.get('post_count', 0),
            member.get('post_points', 0),
            member.get('has_business_license', False),
            member.get('real_name'),
            member.get('phone'),
            member.get('address'),
            bank_accounts_json,
            member.get('alipay'),
            member.get('registration_date'),
            datetime.now()
        ))
        conn.commit()
        cur.close()
        conn.close()
        return True
    except Exception as e:
        print(f'  Save member error: {e}')
        return False
 # ============ 字段解析 ============
 def parse_category(text):
    """识别品类：龙钞、蛇钞、马钞"
    规则：标题或正文中有 龙钞/小龙钞/马钞/蛇钞，或者单独的 龙/马/蛇 字（作为钱币品种）
    """
    if not text:
        return '其他'
    text = str(text)
    # 优先精确匹配
    if any(k in text for k in ['龙钞', '小龙钞']):
        return '龙钞'
    if any(k in text for k in ['蛇钞']):
        return '蛇钞'
    if any(k in text for k in ['马钞', '马年']):
        return '马钞'
    # 单字匹配（作为品种上下文）
    if '龙' in text:
        exclude = ['龙年版', '龙年纪念', '成龙', '恐龙', '龙凤', '天龙', '卧龙', '龙魂', '龙珠笔', '龙马', '龙马精']
        if not any(e in text for e in exclude):
            return '龙钞'
    if '马' in text:
        exclude = ['马自达', '马德里', '马蹄', '马化腾', '马虎', '马蜂窝', '马马虎虎']
        if not any(e in text for e in exclude):
            return '马钞'
    if '蛇' in text:
        exclude = ['眼镜蛇', '蟒蛇', '蛇毒', '捕蛇', '蛇皮']
        if not any(e in text for e in exclude):
            return '蛇钞'
    return '其他'
 def parse_special_types(title):
    """识别特殊规格：标百（刀/刀货）、标十、单张、捆"
    """
    if not title:
        return None
    title = str(title)
    specials = []
    # 标十
    if '标十' in title:
        specials.append('标十')
    # 标百（包含刀、刀货）
    if '标百' in title:
        specials.append('标百')
    elif any(k in title for k in ['刀', '刀货']):
        specials.append('刀')
    # 单张/散张
    if any(k in title for k in ['单张', '散张']):
        specials.append('单张')
    # 捆
    if '捆' in title:
        specials.append('捆')
    # 百连/千连
    if '千连' in title:
        specials.append('千连')
    elif '百连' in title:
        specials.append('百连')
    # 救生圈
    if any(k in title for k in ['大救生圈', '救生圈']):
        specials.append('救生圈')
    return '|'.join(specials) if specials else None
 def parse_number_features(title):
    """识别号码特征：圆圆、倒置、如意、朦胧、金马、金山、天马、钻石、永恒、无4、无47、带4、无347、无247、豹子、狮子、老虎、大象"
    """
    if not title:
        return None
    title = str(title)
    features = []
    feature_map = {
        '倒置': ['倒置'],
        '如意': ['如意'],
        '朦胧': ['朦胧'],
        '金马': ['金马'],
        '金山': ['金山'],
        '天马': ['天马'],
        '钻石': ['钻石'],
        '永恒': ['永恒'],
        '圆圆': ['圆圆'],
        '无4': ['无4', '无四'],
        '无47': ['无47'],
        '无247': ['无247'],
        '无34': ['无34'],
        '无347': ['无347'],
        '带4': ['带4'],
        '豹子': ['豹子'],
        '狮子': ['狮子'],
        '老虎': ['老虎'],
        '大象': ['大象'],
        '生日': ['生日'],
        '满号': ['满号'],
        '首日': ['首日'],
    }
    for feature, keywords in feature_map.items():
        if any(k in title for k in keywords):
            features.append(feature)
    return '|'.join(features) if features else None
 def parse_price(text):
    """从文本解析价格（标准格式用正则，非标准用启发式）
    标准：XXX元/组、XXX/组、XXX元/张、XXX/刀
    非标准：标题中的模糊价格需要结合上下文
    """
    if not text:
        return None, None
    # 优先从标题精确匹配
    patterns = [
        # 格式：数字+元+单位
        (r'(\d{4,5})\s*元\s*/\s*[组张刀条]', '元/组'),
        (r'(\d{4,5})\s*/\s*[组张刀条]', '元/组'),
        (r'(\d{4,5})\s*元', '元'),
        # 求购格式：收XXX元、收XXX
        (r'收\s*(\d{3,5})\s*元?', '元'),
        # 出售格式：出XXX元、售XXX
        (r'[出售售]\s*(\d{3,5})\s*元?', '元'),
    ]
    for pattern, unit in patterns:
        match = re.search(pattern, text)
        if match:
            try:
                price = float(match.group(1))
                # 合理性校验
                if 10 <= price <= 999999:
                    return price, unit
            except:
                pass
    return None, None
 def parse_price_from_content(title, content):
    """从正文提取所有价格，返回最高价格和单位"
    用于多货品帖子，取最高出价/要价
    """
    if not content:
        return parse_price(title)
    # 收集所有符合格式的价格
    prices = []
    price_pattern = r'(\d{3,5})\s*元|收\s*(\d{3,5})|(\d{4,5})\s*/\s*[组张刀]'
    for match in re.finditer(price_pattern, content):
        for group in match.groups():
            if group:
                try:
                    p = float(group)
                    if 10 <= p <= 999999:
                        prices.append(p)
                except:
                    pass
    if prices:
        max_price = max(prices)
        # 确定单位
        unit = '元'
        if '元/组' in content or '/组' in content:
            unit = '元/组'
        elif '元/张' in content or '/张' in content:
            unit = '元/张'
        elif '元/刀' in content or '/刀' in content:
            unit = '元/刀'
        return max_price, unit
    return parse_price(title)
 def parse_post_type(title):
    """识别交易类型：deal(出售)/want(求购)/normal"
    """
    if not title:
        return 'normal'
    title = str(title)
    # 出售关键词
    deal_keywords = ['出售', '转让', '卖', '低价', '低出', '快出', '亏出', '吐血', '清仓', '出 ', '批出', '批售', '甩卖', '特价']
    if any(k in title for k in deal_keywords):
        return 'deal'
    # 求购关键词
    want_keywords = ['求购', '收购', '收 ', '要 ', '收', '求', '要']
    if any(k in title for k in want_keywords):
        return 'want'
    return 'normal'
 # ============ 页面爬取 ============
 def crawl_detail(post_id, url):
    """爬取帖子详情页，提取：内容、用户信息、回复信息"""
    try:
        resp = requests.get(url, headers=HEADERS, timeout=15)
        resp.encoding = 'gb2312'
        html = resp.text
        soup = BeautifulSoup(html, 'html.parser')
        result = {
            'content': None,
            'author_username': None,
            'author_id': None,
            'author_info': {},
            'members': [],
            'reply_count': 0,
            'view_count': 0,
            'post_time': None,
            'contact': None,
            'has_lifebuoy': False
        }
        # 1. 提取浏览数
        view_match = re.search(r'您是本帖的第\s*<b>(\d+)</b>\s*个阅读者', html)
        if view_match:
            result['view_count'] = int(view_match.group(1))
        # 2. 提取回复数（数 postbottom1/2 数量）
        postbottom_count = len(soup.find_all('div', class_=lambda x: x and 'postbottom' in str(x)))
        # 减去1因为第一个是主帖
        result['reply_count'] = max(0, postbottom_count - 1)
        # 3. 解析每个楼层
        page_text = soup.get_text(separator='\n', strip=True)
        # 找所有 postuserinfo 块
        userinfo_blocks = soup.find_all('div', class_='postuserinfo')
        for i, userinfo in enumerate(userinfo_blocks):
            user_data = parse_userinfo_block(userinfo)
            if user_data and user_data.get('user_id'):
                result['members'].append(user_data)
                # 第一条是楼主
                if i == 0:
                    result['author_username'] = user_data.get('username')
                    result['author_id'] = user_data.get('user_id')
                    result['author_info'] = user_data
        # 4. 提取帖子内容
        post_div = soup.find('div', class_='post')
        if post_div:
            text = post_div.get_text(separator='\n', strip=True)
            result['content'] = text[:8000]
            if '救生圈' in text:
                result['has_lifebuoy'] = True
        # 5. 提取联系方式
        phones = re.findall(r'1[3-9]\d{9}', page_text)
        if phones:
            result['contact'] = ','.join(dict.fromkeys(phones[:3]))
        # 6. 提取发帖时间（从第一个postbottom）
        time_div = soup.find('div', class_='postbottom1')
        if time_div:
            time_match = re.search(r'(\d{4})/(\d{1,2})/(\d{1,2})\s+(\d{1,2}):(\d{2})', time_div.get_text())
            if time_match:
                result['post_time'] = f"{time_match.group(1)}-{int(time_match.group(2)):02d}-{int(time_match.group(3)):02d} {time_match.group(4)}:{time_match.group(5)}:00"
        return result
    except Exception as e:
        print(f'  Crawl error: {e}')
        return {}
 def parse_userinfo_block(userinfo_div):
    """解析单个用户的 userinfo div，提取完整用户信息"""
    try:
        text = userinfo_div.get_text(separator='\n', strip=True)
        # user_id - from showyyzz() JS call or j_gbook links
        user_id = None
        userinfo_html = str(userinfo_div)
        js_match = re.search(r'showyyzz\((\d+)\)', userinfo_html)
        if js_match:
            user_id = js_match.group(1)
        else:
            link_match = re.search(r'j_gbook_add\.asp\?id=(\d+)', userinfo_html)
            if link_match:
                user_id = link_match.group(1)
            else:
                dispuser_match = re.search(r'dispuser\.asp\?id=(\d+)', userinfo_html)
                if dispuser_match:
                    user_id = dispuser_match.group(1)
        # username - from <b> tag inside userinfo div
        username = None
        b_tag = userinfo_div.find('b')
        if b_tag:
            username = b_tag.get_text(strip=True)
        # transaction_level
        level_match = re.search(r'交易等级[：:]\s*([^\n]+)', text)
        transaction_level = level_match.group(1).strip() if level_match else None
        # credit_score
        credit_match = re.search(r'信用积分[：:]\s*(\d+)', text)
        credit_score = int(credit_match.group(1)) if credit_match else None
        # rating_count
        rating_match = re.search(r'评分次数[：:]\s*(\d+)', text)
        rating_count = int(rating_match.group(1)) if rating_match else 0
        # post_count
        post_match = re.search(r'发贴次数[：:]\s*(\d+)', text)
        post_count = int(post_match.group(1)) if post_match else 0
        # post_points
        points_match = re.search(r'发帖积分[：:]\s*(\d+)', text)
        post_points = int(points_match.group(1)) if points_match else 0
        # registration_date
        reg_match = re.search(r'注册日期[：:]\s*(\d{4})年(\d{1,2})月(\d{1,2})日', text)
        registration_date = None
        if reg_match:
            try:
                registration_date = f"{reg_match.group(1)}-{int(reg_match.group(2)):02d}-{int(reg_match.group(3)):02d}"
            except:
                pass
        # business license
        has_business_license = '点击查看' in text or '已认证' in text
        # real_name (from 认证员注)
        real_name = None
        name_match = re.search(r'姓名[：:]([^\s\n]+)', text)
        if name_match:
            real_name = name_match.group(1)
        # phone (from userinfo or nearby content)
        phone_match = re.search(r'电话[：:]\s*([^\s\n]+)', text)
        phone = phone_match.group(1).strip() if phone_match else None
        # address
        addr_match = re.search(r'地址[：:]\s*([^\n]+)', text)
        address = addr_match.group(1).strip() if addr_match else None
        # bank_accounts
        bank_accounts = []
        bank_types = ['农行', '工行', '建行', '中行', '交行', '招行', '兴业', '民生', '光大', '中信']
        for bank in bank_types:
            if bank in text:
                # 匹配账号和户名
                matches = re.findall(rf'{bank}[：:]\s*(\d+)\s*([^\s\n]{{2,10}})', text)
                for acc, name in matches:
                    bank_accounts.append({'bank': bank, 'account': acc.strip(), 'name': name.strip()})
        # alipay
        alipay = None
        return {
            'user_id': user_id,
            'username': username,
            'transaction_level': transaction_level,
            'credit_score': credit_score,
            'rating_count': rating_count,
            'post_count': post_count,
            'post_points': post_points,
            'registration_date': registration_date,
            'has_business_license': has_business_license,
            'real_name': real_name,
            'phone': phone,
            'address': address,
            'bank_accounts': bank_accounts,
            'alipay': alipay
        }
    except Exception as e:
        print(f'  Parse userinfo error: {e}')
        return {}
 def crawl_list_page(page):
    """爬取列表页，提取当天所有帖子（使用BeautifulSoup解析）"""
    time.sleep(random.uniform(0.3, 1.5))  # 防封延迟
    url = f'{BASE_URL}/index.asp?boardid={BOARD_ID}&page={page}'
    try:
        resp = requests.get(url, headers=HEADERS, timeout=15)
        resp.encoding = 'gb2312'
        html = resp.text
        soup = BeautifulSoup(html, 'html.parser')
        posts_data = []
        today = datetime.now().strftime('%Y/%-m/%-d')
        # 找到所有包含 boardID=151 链接的 <div class="listtitle">
        listtitle_divs = soup.find_all('div', class_='listtitle')
        for div in listtitle_divs:
            # 在 listtitle div 内找 <a> 标签
            link = div.find('a', href=re.compile(r'dispbbs\.asp\?boardID=151&ID=\d+'))
            if not link:
                continue
            href = link.get('href', '')
            id_match = re.search(r'ID=(\d+)', href)
            if not id_match:
                continue
            post_id = id_match.group(1)
            # 从 title 属性提取信息：格式《标题》\n作者：xxx\n发表于：2026/4/5 9:35:00
            title_attr = link.get('title', '')
            if not title_attr:
                continue
            # 提取日期
            date_match = re.search(r'(\d{4}/\d{1,2}/\d{1,2})', title_attr)
            if not date_match:
                continue
            post_date = date_match.group(1)
            # 只爬当天
            if post_date != today:
                continue
            # 提取标题
            title_match = re.search(r'《([^》]+)》', title_attr)
            title = title_match.group(1).strip() if title_match else ''
            if not title or len(title) < 5:
                continue
            skip_titles = ['栏目交易规范', '固', '精华', '_TOP', '锁', '查看', '页面加载', '投资资讯', '版块主题', '固顶主题']
            if any(t in title for t in skip_titles):
                continue
            full_url = f'{BASE_URL}/dispbbs.asp?boardid={BOARD_ID}&id={post_id}&page=1'
            post = {
                'post_id': post_id,
                'title': title,
                'url': full_url,
                'category': parse_category(title),
                'post_type': parse_post_type(title),
                'special_types': parse_special_types(title),
                'number_features': parse_number_features(title),
            }
            # 从标题提取价格
            price, unit = parse_price(title)
            post['price'] = price
            post['price_unit'] = unit
            post['has_lifebuoy'] = '救生圈' in title
            posts_data.append(post)
        return posts_data
    except Exception as e:
        print(f'Crawl page {page} error: {e}')
        return []
 def process_post(post):
    """处理单个帖子：爬详情 + 提取价格 + 保存会员 + 入库"""
    detail = crawl_detail(post['post_id'], post['url'])
    post.update(detail)
    # 更新作者信息
    if detail.get('author_info'):
        author = detail['author_info']
        post['author_username'] = author.get('username')
        post['author_id'] = author.get('user_id')
    # 保存所有会员
    for member in detail.get('members', []):
        if member.get('user_id'):
            save_member(member)
    # 从正文提取价格（如果标题没有）
    if not post.get('price') and detail.get('content'):
        price, unit = parse_price_from_content(post.get('title', ''), detail.get('content', ''))
        if price:
            post['price'] = price
            post['price_unit'] = unit
    # 合并 content
    if detail.get('content'):
        post['content'] = detail['content']
    if save_post(post):
        return 1
    return 0
 def run(max_pages=5, workers=10):
    spider_name = 'yichens_spider_v4'
    start_time = datetime.now()
    print('=' * 60)
    print(f'一尘网爬虫 v4 [{spider_name}]')
    print(f'时间: {start_time.strftime("%Y-%m-%d %H:%M:%S")}')
    print(f'并发: {workers} 线程')
    print('=' * 60)
    # 1. 已有post_id（增量去重）
    existing_ids = get_existing_post_ids()
    print(f'数据库已有帖子: {len(existing_ids)} 条')
    # 2. 爬列表页
    all_posts = []
    for page in range(1, max_pages + 1):
        print(f'扫描第 {page} 页...')
        posts = crawl_list_page(page)
        if not posts:
            if page > 2:
                print(f'第 {page} 页无新数据，停止')
                break
        else:
            all_posts.extend(posts)
            print(f'  第 {page} 页: {len(posts)} 条')
    # 3. 增量过滤
    new_posts = [p for p in all_posts if p['post_id'] not in existing_ids]
    print(f'\n共找到 {len(all_posts)} 条今日帖子，新增 {len(new_posts)} 条')
    if not new_posts:
        print('没有新帖子，退出')
        log_crawl(spider_name, 'success', 0, '')
        return 0
    # 4. 并发处理
    saved = 0
    member_count = 0
    print(f'开始并发爬取 ({workers} 线程)...')
    with ThreadPoolExecutor(max_workers=workers) as executor:
        future_to_post = {executor.submit(process_post, post): post for post in new_posts}
        for i, future in enumerate(as_completed(future_to_post)):
            post = future_to_post[future]
            try:
                if future.result():
                    saved += 1
                    # 统计会员数（从 post 对象获取）
                    if post.get('author_id'):
                        member_count += 1
                if (i + 1) % 20 == 0:
                    print(f'进度: {i+1}/{len(new_posts)}, 已保存: {saved}')
            except Exception as e:
                print(f'处理 {post["post_id"]} 异常: {e}')
    elapsed = (datetime.now() - start_time).total_seconds()
    print(f'\n完成! 新增帖子 {saved} 条, 会员 {member_count} 人, 耗时 {elapsed:.1f}秒')
    log_crawl(spider_name, 'success', saved, '')
    return saved
 if __name__ == '__main__':
    run(max_pages=5, workers=10)
--- a/database.py
+++ b/database.py
@ -1,105 +1,57 @@
-"""PostgreSQL database connection module"""
+"""MySQL 数据库连接池模块"""
-import psycopg2
+import mysql.connector
-import os
+from mysql.connector import pooling
 from contextlib import contextmanager
-DB_CONFIG = {
+class Database:
-    "host": os.environ.get("DB_HOST", "pgm-bp1t1008h019ez6c.pg.rds.aliyuncs.com"),
+    _pool = None
-    "port": int(os.environ.get("DB_PORT", 5432)),
+    
-    "user": os.environ.get("DB_USER", "coolbot"),
+    @classmethod
-    "password": os.environ.get("DB_PASSWORD", "Coolbot123"),
+    def get_pool(cls):
-    "database": os.environ.get("DB_NAME", "coolbot_data"),
+        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_db():
+    def get_connection(self):
-    conn = None
+        conn = self.get_pool().get_connection()
        try:
        conn = psycopg2.connect(**DB_CONFIG)
            yield conn
        conn.commit()
    except Exception as e:
        if conn:
            conn.rollback()
        raise e
        finally:
        if conn:
            conn.close()
 class DictCursor:
    def __init__(self, cursor):
        self._cursor = cursor
    def __iter__(self):
        columns = [desc[0] for desc in self._cursor.description] if self._cursor.description else []
        for row in self._cursor:
            yield dict(zip(columns, row))
    def __enter__(self):
        return self
    def __exit__(self, *args):
        self.close()
    def execute(self, *args, **kwargs):
        return self._cursor.execute(*args, **kwargs)
    def fetchone(self):
        row = self._cursor.fetchone()
        if row is None:
            return None
        columns = [desc[0] for desc in self._cursor.description] if self._cursor.description else []
        return dict(zip(columns, row))
    def fetchall(self):
        columns = [desc[0] for desc in self._cursor.description] if self._cursor.description else []
        return [dict(zip(columns, row)) for row in self._cursor.fetchall()]
    def fetchmany(self, size=None):
        if size is None:
            size = self._cursor.arraysize
        columns = [desc[0] for desc in self._cursor.description] if self._cursor.description else []
        rows = self._cursor.fetchmany(size)
        return [dict(zip(columns, row)) for row in rows]
    def close(self):
        return self._cursor.close()
 class Database:
    @contextmanager
    def get_cursor(self, dictionary=True):
-        conn = None
+        with self.get_connection() as conn:
            cursor = conn.cursor(dictionary=dictionary)
            try:
            conn = psycopg2.connect(**DB_CONFIG)
            raw_cursor = conn.cursor()
            try:
                if dictionary:
                    cursor = DictCursor(raw_cursor)
                else:
                    cursor = raw_cursor
                yield cursor
                conn.commit()
            except Exception as e:
                conn.rollback()
                raise e
            finally:
-                raw_cursor.close()
+                cursor.close()
        finally:
            if conn:
                conn.close()
 db = Database()
 def test_connection():
    """测试数据库连接"""
    try:
-        with get_db() as conn:
+        with db.get_cursor() as cursor:
-            cur = conn.cursor()
+            cursor.execute("SELECT 1 as test")
-            cur.execute("SELECT version()")
+            result = cursor.fetchone()
-            print(f"DB OK: {str(cur.fetchone()[0])[:50]}")
+            print(f"✅ 数据库连接成功: {result}")
            return True
    except Exception as e:
-        print(f"DB ERROR: {e}")
+        print(f"❌ 数据库连接失败: {e}")
        return False
 if __name__ == "__main__":
    test_connection()
--- a/llm_price_extract.py
+++ b/llm_price_extract.py
@ -1,107 +0,0 @@
 #!/usr/bin/env python3
 """LLM-based price extraction for post titles"""
 import os, json, re
 import httpx
 LLM_API_KEY = os.environ.get('LLM_API_KEY', 'sk-sp-d5ce68bb203e48ca857c2aea25255b26')
 LLM_API_URL = os.environ.get('LLM_API_URL', 'https://coding.dashscope.aliyuncs.com/v1/chat/completions')
 LLM_MODEL = os.environ.get('LLM_MODEL', 'qwen3.5-plus')
 def extract_price_with_llm(title: str) -> dict:
    """Use LLM to extract price from title. Returns dict with price, unit, confidence."""
    prompt = f"""你是一个钱币收藏市场的价格分析师。请从以下帖子标题中提取信息：
 标题："{title}"
 请仔细分析：
 1. 这个帖子是求购还是出售？（收/求/购 = 求购，出/售 = 出售）
 2. 实际交易价格是多少？（数字+单位）
 3. 价格单位是什么？（元/张、元/刀、元/条、元/套 等）
 注意：
 - "求2组"不是价格，"2组"只是数量
 - "3月"不是价格，是日期
 - 只有明确表示交易价格的才是价格
 请用JSON格式回答：{{"post_type":"want/deal/null","price":数字或null,"unit":"元/张等","reason":"解释"}}
 只回答JSON，不要其他内容。"""
    try:
        with httpx.Client(timeout=30.0) as client:
            response = client.post(
                LLM_API_URL,
                headers={
                    "Authorization": f"Bearer {LLM_API_KEY}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": LLM_MODEL,
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.1
                }
            )
            if response.status_code == 200:
                result = response.json()
                content = result['choices'][0]['message']['content'].strip()
                # Extract JSON
                if '{' in content:
                    json_str = content[content.find('{'):content.rfind('}')+1]
                    return json.loads(json_str)
    except Exception as e:
        print(f"LLM error: {e}")
    return {"post_type": None, "price": None, "unit": None, "reason": "LLM failed"}
 def batch_extract_prices(titles: list) -> list:
    """Batch extract prices from multiple titles"""
    prompt = f"""你是一个钱币收藏市场的价格分析师。请批量分析以下帖子标题，提取求购/出售价格信息。
 标题列表：
 {chr(10).join([f"{i+1}. {t}" for i, t in enumerate(titles)])}
 对于每个标题，判断：
 - post_type: "want"表示求购，"deal"表示出售，"null"表示无法判断
 - price: 实际交易价格数字（元），如果不是价格或无法判断则填null
 - unit: 价格单位，如"元/张"、"元/刀"、"元/条"、"元/套"、"元"等
 只返回JSON数组格式：[{{"idx":1,"post_type":"want","price":120,"unit":"元/张","reason":"..."}},...]
 只回答JSON数组。"""
    try:
        with httpx.Client(timeout=60.0) as client:
            response = client.post(
                LLM_API_URL,
                headers={
                    "Authorization": f"Bearer {LLM_API_KEY}",
                    "Content-Type": "application/json"
                },
                json={
                    "model": LLM_MODEL,
                    "messages": [{"role": "user", "content": prompt}],
                    "temperature": 0.1
                }
            )
            if response.status_code == 200:
                result = response.json()
                content = result['choices'][0]['message']['content'].strip()
                if '[' in content:
                    json_str = content[content.find('['):content.rfind(']')+1]
                    return json.loads(json_str)
    except Exception as e:
        print(f"Batch LLM error: {e}")
    return []
 if __name__ == '__main__':
    test_titles = [
        "求2组小龙鈔无四七标十，爱藏67+三星",
        "765出一组无三四七标十马钞三包到手 可小义",
        "2000元出一组龙钞朦胧号5张PMG68分",
        "收购龙钞带4标十 1200元/张",
        "低价出蛇钞一刀 已经刀切好",
        "求购小龙钞无47标十 450元每张",
    ]
    print("Testing LLM price extraction:")
    for title in test_titles:
        result = extract_price_with_llm(title)
        print(f"\n标题: {title}")
        print(f"结果: {result}")
--- a/requirements.txt
+++ b/requirements.txt
@ -6,11 +6,13 @@ uvicorn[standard]>=0.27.0
 pydantic>=2.5.0
 # Database
-psycopg2-binary>=2.9.9
+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
--- a/scheduler.py
+++ b/scheduler.py
@ -1,73 +0,0 @@
 """
 定时任务调度器 - 修复版
 集成 APScheduler 实现爬虫自动化
 """
 import logging
 from apscheduler.schedulers.blocking import BlockingScheduler
 from apscheduler.triggers.cron import CronTrigger
 from apscheduler.triggers.interval import IntervalTrigger
 logger = logging.getLogger(__name__)
 # 全局调度器实例
 scheduler = BlockingScheduler(timezone="Asia/Shanghai")
 def crawl_today_job():
    """每日采集任务"""
    from crawlers.crawl_today import YichensTodaySpider
    logger.info("[定时任务] 开始执行一尘网今日采集")
    try:
        spider = YichensTodaySpider()
        result = spider.run()
        logger.info(f"[定时任务] 采集完成，新增 {result} 条")
    except Exception as e:
        logger.error(f"[定时任务] 采集失败: {e}")
 def crawl_incremental_job():
    """增量采集任务（每小时）"""
    from crawlers.crawl_today import YichensTodaySpider
    logger.info("[定时任务] 开始执行增量采集")
    try:
        spider = YichensTodaySpider()
        spider.max_pages = 2  # 增量只扫2页
        result = spider.crawl_today()
        logger.info(f"[定时任务] 增量采集完成，新增 {result} 条")
    except Exception as e:
        logger.error(f"[定时任务] 增量采集失败: {e}")
 def init_scheduler():
    """初始化定时任务"""
    # 每天早上 8 点采集
    scheduler.add_job(
        crawl_today_job,
        CronTrigger(hour=8, minute=0, timezone="Asia/Shanghai"),
        id='yichens_daily_crawl',
        name='一尘网每日采集',
        replace_existing=True
    )
    # 每小时增量采集（如果需要）
    scheduler.add_job(
        crawl_incremental_job,
        IntervalTrigger(hours=1, timezone="Asia/Shanghai"),
        id='yichens_incremental',
        name='一尘网增量采集',
        replace_existing=True
    )
    logger.info("定时任务已注册: 每日8点采集 + 每小时增量")
    return scheduler
 def start_scheduler():
    """启动调度器"""
    init_scheduler()
    logger.info("调度器启动")
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        logger.info("调度器停止")
        scheduler.shutdown()
 if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    start_scheduler()
--- a/scripts/crawl_cron.sh
+++ b/scripts/crawl_cron.sh
@ -1,19 +0,0 @@
 #!/bin/bash
 cd /root/coolbot-data
 source venv/bin/activate
 python3 -u -c "
 import sys, os
 sys.path.insert(0, '/root/coolbot-data')
 os.environ['DB_HOST'] = 'pgm-bp1t1008h019ez6c.pg.rds.aliyuncs.com'
 os.environ['DB_PORT'] = '5432'
 os.environ['DB_USER'] = 'coolbot'
 os.environ['DB_PASSWORD'] = 'Coolbot123'
 os.environ['DB_NAME'] = 'coolbot_data'
 from crawlers.crawl_today import YichensTodaySpider
 spider = YichensTodaySpider()
 spider.max_pages = 2
 spider.min_delay = 0.2
 spider.max_delay = 0.5
 result = spider.crawl_today()
 print(f'采集完成: {result} 条', flush=True)
 "
--- a/scripts/daily_report.py
+++ b/scripts/daily_report.py
@ -27,7 +27,7 @@ def get_yichens_stats():
        total_posts = cursor.fetchone()["cnt"]
        # 今日新增
-        cursor.execute("SELECT COUNT(*) as cnt FROM yichens_posts WHERE DATE(crawled_at) = CURRENT_DATE")
+        cursor.execute("SELECT COUNT(*) as cnt FROM yichens_posts WHERE DATE(crawled_at) = CURDATE()")
        today_posts = cursor.fetchone()["cnt"]
        # 交易帖数量
@ -46,7 +46,7 @@ def get_yichens_stats():
                MIN(price) as min_price,
                MAX(price) as max_price
            FROM yichens_posts 
-            WHERE price IS NOT NULL AND price > 0 AND DATE(crawled_at) = CURRENT_DATE
+            WHERE price IS NOT NULL AND price > 0 AND DATE(crawled_at) = CURDATE()
        """)
        today_price_stats = cursor.fetchone()