Compare commits
2 Commits
07b2e5c7b5
...
c330d3f0c2
| Author | SHA1 | Date |
|---|---|---|
caibotmini
|
c330d3f0c2 | |
|
caibotmini
|
9a6f0de418 |
22
.env.example
22
.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
|
||||
120
README.md
120
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
|
||||
```
|
||||
酷博特数据分析系统 - 多源数据采集、存储、清洗、分析全链路自动化平台
|
||||
41
alembic.ini
41
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
|
||||
|
|
@ -1 +0,0 @@
|
|||
Generic single-database configuration.
|
||||
152
alembic/env.py
152
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()
|
||||
|
|
@ -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"}
|
||||
|
|
@ -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
|
||||
|
|
@ -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"]
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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':
|
||||
where_clauses.append("post_type NOT IN ('deal','want')")
|
||||
elif post_type:
|
||||
if post_type:
|
||||
where_clauses.append("post_type = %s")
|
||||
params.append(post_type)
|
||||
if author_username:
|
||||
where_clauses.append("author_username LIKE %s")
|
||||
params.append(f"%{author_username}%")
|
||||
if keyword:
|
||||
where_clauses.append("(title ILIKE %s OR content ILIKE %s)")
|
||||
params.extend([f"%{keyword}%", f"%{keyword}%"])
|
||||
where_clauses.append("MATCH(title, content) AGAINST(%s IN NATURAL LANGUAGE MODE)")
|
||||
params.append(keyword)
|
||||
if min_price is not None:
|
||||
where_clauses.append("price >= %s")
|
||||
params.append(min_price)
|
||||
if max_price is not None:
|
||||
where_clauses.append("price <= %s")
|
||||
params.append(max_price)
|
||||
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:
|
||||
cur = conn.cursor()
|
||||
cur.execute(f"SELECT COUNT(*) as total FROM yichens_posts{where_sql}", params)
|
||||
total = cur.fetchone()[0]
|
||||
cur.execute(f"SELECT * FROM yichens_posts{where_sql} ORDER BY post_time DESC LIMIT %s OFFSET %s", params + [page_size, offset])
|
||||
rows = cur.fetchall()
|
||||
cols = [desc[0] for desc in cur.description]
|
||||
items = [dict(zip(cols, row)) for row in rows]
|
||||
cur.close()
|
||||
with db.get_cursor() as cursor:
|
||||
cursor.execute(f"SELECT COUNT(*) as total FROM yichens_posts{where_sql}", params)
|
||||
total = cursor.fetchone()["total"]
|
||||
|
||||
cursor.execute(f"""
|
||||
SELECT * FROM yichens_posts{where_sql}
|
||||
ORDER BY created_at DESC LIMIT %s OFFSET %s
|
||||
""", params + [page_size, offset])
|
||||
items = cursor.fetchall()
|
||||
|
||||
return ApiResponse.success(PaginatedData.create(items, page, page_size, total))
|
||||
|
||||
|
||||
@router.get("/posts/{post_id}", response_model=ApiResponse)
|
||||
async def get_yichens_post(post_id: str):
|
||||
"""获取一尘帖子详情"""
|
||||
|
|
@ -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})
|
||||
|
||||
|
|
|
|||
134
cache.py
134
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()
|
||||
|
|
@ -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} 条')
|
||||
|
|
@ -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(' ', ' ').replace(' ', ' ')
|
||||
text = text.replace('&', '&').replace('<', '<').replace('>', '>')
|
||||
|
||||
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()
|
||||
|
|
@ -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_with_unit = [
|
||||
r'(\d+(?:\.\d+)?)\s*[万Ww]?\s*元', # 123元, 1.5万元
|
||||
r'[每单共总]\s*(\d+(?:\.\d+)?)\s*元', # 共123元
|
||||
r'(\d+(?:\.\d+)?)\s*元\s*(?:每|每张|每刀|每条|每套)', # 123元每张
|
||||
patterns = [
|
||||
r'(\d+)\s*[万Ww]?\s*元',
|
||||
r'(\d+)\s*-\s*(\d+)\s*[万Ww]?',
|
||||
r'[收售出求兑买]\s*(\d+)',
|
||||
]
|
||||
for pattern in patterns_with_unit:
|
||||
for pattern in patterns:
|
||||
m = re.search(pattern, text)
|
||||
if m:
|
||||
try:
|
||||
val = float(m.group(1))
|
||||
if val > 0:
|
||||
return val
|
||||
return float(m.group(1))
|
||||
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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
110
database.py
110
database.py
|
|
@ -1,105 +1,57 @@
|
|||
"""PostgreSQL database connection module"""
|
||||
import psycopg2
|
||||
import os
|
||||
"""MySQL 数据库连接池模块"""
|
||||
import mysql.connector
|
||||
from mysql.connector import pooling
|
||||
from contextlib import contextmanager
|
||||
|
||||
DB_CONFIG = {
|
||||
"host": os.environ.get("DB_HOST", "pgm-bp1t1008h019ez6c.pg.rds.aliyuncs.com"),
|
||||
"port": int(os.environ.get("DB_PORT", 5432)),
|
||||
"user": os.environ.get("DB_USER", "coolbot"),
|
||||
"password": os.environ.get("DB_PASSWORD", "Coolbot123"),
|
||||
"database": os.environ.get("DB_NAME", "coolbot_data"),
|
||||
}
|
||||
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_db():
|
||||
conn = None
|
||||
def get_connection(self):
|
||||
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()
|
||||
finally:
|
||||
if conn:
|
||||
conn.close()
|
||||
cursor.close()
|
||||
|
||||
db = Database()
|
||||
|
||||
def test_connection():
|
||||
"""测试数据库连接"""
|
||||
try:
|
||||
with get_db() as conn:
|
||||
cur = conn.cursor()
|
||||
cur.execute("SELECT version()")
|
||||
print(f"DB OK: {str(cur.fetchone()[0])[:50]}")
|
||||
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"DB ERROR: {e}")
|
||||
print(f"❌ 数据库连接失败: {e}")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_connection()
|
||||
|
|
|
|||
|
|
@ -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}")
|
||||
|
|
@ -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
|
||||
|
|
|
|||
73
scheduler.py
73
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()
|
||||
|
|
@ -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)
|
||||
"
|
||||
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue