Compare commits
4 Commits
c330d3f0c2
...
07b2e5c7b5
| Author | SHA1 | Date |
|---|---|---|
|
|
07b2e5c7b5 | |
|
|
76d87adadb | |
|
|
4282eac809 | |
|
|
9d418f3d8c |
|
|
@ -0,0 +1,22 @@
|
|||
# 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,3 +1,119 @@
|
|||
# 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
|
||||
```
|
||||
|
|
|
|||
|
|
@ -0,0 +1,41 @@
|
|||
[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
|
||||
|
|
@ -0,0 +1 @@
|
|||
Generic single-database configuration.
|
||||
|
|
@ -0,0 +1,152 @@
|
|||
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()
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
"""${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"}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
"""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) = CURDATE()
|
||||
WHERE DATE(crawled_at) = CURRENT_DATE
|
||||
""")
|
||||
today_price_records = cursor.fetchone()["cnt"]
|
||||
|
||||
|
|
|
|||
|
|
@ -31,6 +31,10 @@ 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
|
||||
|
|
@ -38,6 +42,10 @@ 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
|
||||
from database import db, get_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=100)
|
||||
page_size: int = Query(20, ge=1, le=500)
|
||||
):
|
||||
"""获取一尘用户列表"""
|
||||
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()["total"]
|
||||
total = cursor.fetchone()[0]
|
||||
|
||||
cursor.execute(f"""
|
||||
SELECT * FROM yichens_users{where_sql}
|
||||
|
|
@ -110,47 +110,58 @@ 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=100)
|
||||
page_size: int = Query(20, ge=1, le=500)
|
||||
):
|
||||
"""获取一尘帖子列表"""
|
||||
offset = (page - 1) * page_size
|
||||
where_clauses = []
|
||||
where_clauses = ["1=1"]
|
||||
params = []
|
||||
|
||||
if category:
|
||||
where_clauses.append("category = %s")
|
||||
params.append(category)
|
||||
if post_type:
|
||||
if post_type == 'other':
|
||||
where_clauses.append("post_type NOT IN ('deal','want')")
|
||||
elif post_type:
|
||||
where_clauses.append("post_type = %s")
|
||||
params.append(post_type)
|
||||
if author_username:
|
||||
where_clauses.append("author_username LIKE %s")
|
||||
params.append(f"%{author_username}%")
|
||||
if keyword:
|
||||
where_clauses.append("MATCH(title, content) AGAINST(%s IN NATURAL LANGUAGE MODE)")
|
||||
params.append(keyword)
|
||||
where_clauses.append("(title ILIKE %s OR content ILIKE %s)")
|
||||
params.extend([f"%{keyword}%", f"%{keyword}%"])
|
||||
if min_price is not None:
|
||||
where_clauses.append("price >= %s")
|
||||
params.append(min_price)
|
||||
if max_price is not None:
|
||||
where_clauses.append("price <= %s")
|
||||
params.append(max_price)
|
||||
if time_range == '1h':
|
||||
where_clauses.append("post_time >= NOW() - INTERVAL '1 hour'")
|
||||
elif time_range == '6h':
|
||||
where_clauses.append("post_time >= NOW() - INTERVAL '6 hours'")
|
||||
elif time_range == '24h':
|
||||
where_clauses.append("post_time >= NOW() - INTERVAL '24 hours'")
|
||||
elif time_range == 'today':
|
||||
where_clauses.append("post_time::date = CURRENT_DATE")
|
||||
|
||||
where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else ""
|
||||
where_sql = " WHERE " + " AND ".join(where_clauses)
|
||||
|
||||
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()
|
||||
with get_db() as conn:
|
||||
cur = conn.cursor()
|
||||
cur.execute(f"SELECT COUNT(*) as total FROM yichens_posts{where_sql}", params)
|
||||
total = cur.fetchone()[0]
|
||||
cur.execute(f"SELECT * FROM yichens_posts{where_sql} ORDER BY post_time DESC LIMIT %s OFFSET %s", params + [page_size, offset])
|
||||
rows = cur.fetchall()
|
||||
cols = [desc[0] for desc in cur.description]
|
||||
items = [dict(zip(cols, row)) for row in rows]
|
||||
cur.close()
|
||||
|
||||
return ApiResponse.success(PaginatedData.create(items, page, page_size, total))
|
||||
|
||||
|
||||
@router.get("/posts/{post_id}", response_model=ApiResponse)
|
||||
async def get_yichens_post(post_id: str):
|
||||
"""获取一尘帖子详情"""
|
||||
|
|
@ -236,7 +247,7 @@ async def get_post_replies(
|
|||
"SELECT COUNT(*) as total FROM yichens_replies WHERE post_id = %s",
|
||||
(post_id,)
|
||||
)
|
||||
total = cursor.fetchone()["total"]
|
||||
total = cursor.fetchone()[0]
|
||||
|
||||
cursor.execute("""
|
||||
SELECT * FROM yichens_replies
|
||||
|
|
@ -309,16 +320,206 @@ async def get_yichens_statistics():
|
|||
cursor.execute("""
|
||||
SELECT DATE(crawled_at) as date, COUNT(*) as count
|
||||
FROM yichens_posts
|
||||
WHERE crawled_at >= DATE_SUB(NOW(), INTERVAL 7 DAY)
|
||||
WHERE crawled_at >= NOW() - INTERVAL '7 days'
|
||||
GROUP BY DATE(crawled_at)
|
||||
ORDER BY date DESC
|
||||
""")
|
||||
daily_stats = cursor.fetchall()
|
||||
|
||||
# Add today counts
|
||||
cursor.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_time::date = CURRENT_DATE")
|
||||
today_count = cursor.fetchone()["count"]
|
||||
cursor.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'deal' AND post_time::date = CURRENT_DATE")
|
||||
deal_count = cursor.fetchone()["count"]
|
||||
cursor.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'want' AND post_time::date = CURRENT_DATE")
|
||||
want_count = cursor.fetchone()["count"]
|
||||
cursor.execute("SELECT COUNT(*) FROM yichens_posts WHERE has_lifebuoy = TRUE AND post_time::date = CURRENT_DATE")
|
||||
lifebuoy_count = cursor.fetchone()["count"]
|
||||
|
||||
return ApiResponse.success({
|
||||
"total_users": total_users,
|
||||
"total_posts": total_posts,
|
||||
"today_count": today_count,
|
||||
"deal_count": deal_count,
|
||||
"want_count": want_count,
|
||||
"lifebuoy_count": lifebuoy_count,
|
||||
"seller_count": seller_count,
|
||||
"category_stats": category_stats,
|
||||
"daily_stats": daily_stats
|
||||
})
|
||||
|
||||
@router.get("/report/categories", response_model=ApiResponse)
|
||||
async def get_category_report():
|
||||
with get_db() as conn:
|
||||
cur = conn.cursor()
|
||||
cur.execute("""
|
||||
SELECT COALESCE(category, '其他') as category, COUNT(*) as cnt
|
||||
FROM yichens_posts
|
||||
WHERE post_time::date = CURRENT_DATE
|
||||
GROUP BY COALESCE(category, '其他')
|
||||
ORDER BY cnt DESC
|
||||
""")
|
||||
today_cats = [{"category": r[0], "count": r[1]} for r in cur.fetchall()]
|
||||
cur.execute("""
|
||||
SELECT COALESCE(category, '其他') as category, COUNT(*) as cnt
|
||||
FROM yichens_posts
|
||||
WHERE post_time >= NOW() - INTERVAL '1 hour'
|
||||
GROUP BY COALESCE(category, '其他')
|
||||
ORDER BY cnt DESC
|
||||
""")
|
||||
last1h_cats = [{"category": r[0], "count": r[1]} for r in cur.fetchall()]
|
||||
cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_time::date = CURRENT_DATE")
|
||||
today_total = cur.fetchone()[0]
|
||||
cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_time >= NOW() - INTERVAL '1 hour'")
|
||||
last1h_total = cur.fetchone()[0]
|
||||
cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'deal' AND post_time::date = CURRENT_DATE")
|
||||
today_deal = cur.fetchone()[0]
|
||||
cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'want' AND post_time::date = CURRENT_DATE")
|
||||
today_want = cur.fetchone()[0]
|
||||
cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE has_lifebuoy = TRUE AND post_time::date = CURRENT_DATE")
|
||||
today_lifebuoy = cur.fetchone()[0]
|
||||
cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'deal' AND post_time >= NOW() - INTERVAL '1 hour'")
|
||||
last1h_deal = cur.fetchone()[0]
|
||||
cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE post_type = 'want' AND post_time >= NOW() - INTERVAL '1 hour'")
|
||||
last1h_want = cur.fetchone()[0]
|
||||
cur.execute("SELECT COUNT(*) FROM yichens_posts WHERE has_lifebuoy = TRUE AND post_time >= NOW() - INTERVAL '1 hour'")
|
||||
last1h_lifebuoy = cur.fetchone()[0]
|
||||
cur.close()
|
||||
return ApiResponse.success({
|
||||
"today": {"total": today_total, "deal": today_deal, "want": today_want, "lifebuoy": today_lifebuoy, "categories": today_cats},
|
||||
"last_1h": {"total": last1h_total, "deal": last1h_deal, "want": last1h_want, "lifebuoy": last1h_lifebuoy, "categories": last1h_cats}
|
||||
})
|
||||
|
||||
@router.get("/report/stats", response_model=ApiResponse)
|
||||
async def get_longtian_stats():
|
||||
"""龙钞统计分析"""
|
||||
with get_db() as conn:
|
||||
cur = conn.cursor()
|
||||
|
||||
# 收购关键词
|
||||
want_kw = '%(龙|龙钞|小龙钞)%'
|
||||
want_cond = "(title ILIKE '%龙%' OR title ILIKE '%龙钞%' OR title ILIKE '%小龙钞%') AND (title ILIKE '%收%' OR title ILIKE '%求%' OR title ILIKE '%购%')"
|
||||
# 出售关键词
|
||||
deal_kw = '%(龙|龙钞|小龙钞)%'
|
||||
deal_cond = "(title ILIKE '%龙%' OR title ILIKE '%龙钞%' OR title ILIKE '%小龙钞%') AND (title ILIKE '%出%' OR title ILIKE '%售%')"
|
||||
|
||||
def count_by_pattern(cur, base_cond, extra_kw):
|
||||
"""统计满足基本条件+特殊关键词的帖子数量"""
|
||||
where = base_cond + " AND " + extra_kw + " AND post_time::date = CURRENT_DATE"
|
||||
cur.execute(f"SELECT COUNT(*) FROM yichens_posts WHERE {where}")
|
||||
return cur.fetchone()[0]
|
||||
|
||||
def avg_price_by_pattern(cur, base_cond, extra_kw):
|
||||
"""统计满足基本条件+特殊关键词的帖子平均价格"""
|
||||
where = base_cond + " AND " + extra_kw + " AND post_time::date = CURRENT_DATE AND price IS NOT NULL AND price > 0"
|
||||
cur.execute(f"SELECT COALESCE(AVG(price), 0), COUNT(*) FROM yichens_posts WHERE {where}")
|
||||
r = cur.fetchone()
|
||||
return {"avg": int(r[0]) if r[0] else 0, "count": r[1]}
|
||||
|
||||
# 收购类 - 带4标十
|
||||
c1 = count_by_pattern(cur, want_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'")
|
||||
p1 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'")
|
||||
|
||||
# 收购类 - 无47标十
|
||||
c2 = count_by_pattern(cur, want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'")
|
||||
p2 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'")
|
||||
|
||||
# 收购类 - 无247标十
|
||||
c3 = count_by_pattern(cur, want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'")
|
||||
p3 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'")
|
||||
|
||||
# 收购类 - 无247标百
|
||||
c4 = count_by_pattern(cur, want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')")
|
||||
p4 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')")
|
||||
|
||||
# 收购类 - 无47标百
|
||||
c5 = count_by_pattern(cur, want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')")
|
||||
p5 = avg_price_by_pattern(cur, want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')")
|
||||
|
||||
# 出售类 - 带4标十
|
||||
c6 = count_by_pattern(cur, deal_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'")
|
||||
p6 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'")
|
||||
|
||||
# 出售类 - 无47标十
|
||||
c7 = count_by_pattern(cur, deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'")
|
||||
p7 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'")
|
||||
|
||||
# 出售类 - 无247标十
|
||||
c8 = count_by_pattern(cur, deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'")
|
||||
p8 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'")
|
||||
|
||||
# 出售类 - 无247标百
|
||||
c9 = count_by_pattern(cur, deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')")
|
||||
p9 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')")
|
||||
|
||||
# 出售类 - 无47标百
|
||||
c10 = count_by_pattern(cur, deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')")
|
||||
p10 = avg_price_by_pattern(cur, deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')")
|
||||
|
||||
cur.close()
|
||||
|
||||
return ApiResponse.success({
|
||||
"want": {
|
||||
"dai4_biao10": {"count": c1, **p1},
|
||||
"wu47_biao10": {"count": c2, **p2},
|
||||
"wu247_biao10": {"count": c3, **p3},
|
||||
"wu247_biaobai": {"count": c4, **p4},
|
||||
"wu47_biaobai": {"count": c5, **p5},
|
||||
},
|
||||
"deal": {
|
||||
"dai4_biao10": {"count": c6, **p6},
|
||||
"wu47_biao10": {"count": c7, **p7},
|
||||
"wu247_biao10": {"count": c8, **p8},
|
||||
"wu247_biaobai": {"count": c9, **p9},
|
||||
"wu47_biaobai": {"count": c10, **p10},
|
||||
}
|
||||
})
|
||||
|
||||
@router.get("/report/posts", response_model=ApiResponse)
|
||||
async def get_report_posts(
|
||||
stat_type: str = Query(..., description="统计类型: want_dai4_biao10, want_wu47_biao10, deal_dai4_biao10, etc"),
|
||||
page: int = Query(1, ge=1),
|
||||
page_size: int = Query(20, ge=1, le=500)
|
||||
):
|
||||
"""获取统计报表对应的帖子列表"""
|
||||
offset = (page - 1) * page_size
|
||||
|
||||
# 基础条件:龙钞类
|
||||
longtian_cond = "(title ILIKE '%龙%' OR title ILIKE '%龙钞%' OR title ILIKE '%小龙钞%')"
|
||||
|
||||
# 收购/出售条件
|
||||
want_cond = longtian_cond + " AND (title ILIKE '%收%' OR title ILIKE '%求%' OR title ILIKE '%购%')"
|
||||
deal_cond = longtian_cond + " AND (title ILIKE '%出%' OR title ILIKE '%售%')"
|
||||
|
||||
# stat_type 到 WHERE 条件的映射
|
||||
stat_map = {
|
||||
"want_dai4_biao10": (want_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'"),
|
||||
"want_wu47_biao10": (want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'"),
|
||||
"want_wu247_biao10": (want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'"),
|
||||
"want_wu247_biaobai": (want_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')"),
|
||||
"want_wu47_biaobai": (want_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')"),
|
||||
"deal_dai4_biao10": (deal_cond, "(title ILIKE '%带4%' OR title ILIKE '%带四%' OR title ILIKE '%标配%') AND title ILIKE '%标十%'"),
|
||||
"deal_wu47_biao10": (deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND title ILIKE '%标十%'"),
|
||||
"deal_wu247_biao10": (deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND title ILIKE '%标十%'"),
|
||||
"deal_wu247_biaobai": (deal_cond, "(title ILIKE '%无247%' OR title ILIKE '%天马%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')"),
|
||||
"deal_wu47_biaobai": (deal_cond, "(title ILIKE '%无47%' OR title ILIKE '%无四七%') AND (title ILIKE '%标百%' OR title ILIKE '%刀%')"),
|
||||
}
|
||||
|
||||
if stat_type not in stat_map:
|
||||
return ApiResponse.success({"items": [], "total": 0, "page": page, "page_size": page_size})
|
||||
|
||||
base_cond, extra_cond = stat_map[stat_type]
|
||||
where_sql = f" WHERE {base_cond} AND {extra_cond} AND post_time::date = CURRENT_DATE"
|
||||
|
||||
with get_db() as conn:
|
||||
cur = conn.cursor()
|
||||
cur.execute(f"SELECT COUNT(*) FROM yichens_posts{where_sql}")
|
||||
total = cur.fetchone()[0]
|
||||
cur.execute(f"SELECT * FROM yichens_posts{where_sql} ORDER BY post_time DESC LIMIT {page_size} OFFSET {offset}")
|
||||
rows = cur.fetchall()
|
||||
cols = [desc[0] for desc in cur.description]
|
||||
items = [dict(zip(cols, row)) for row in rows]
|
||||
cur.close()
|
||||
|
||||
return ApiResponse.success({"items": items, "total": total, "page": page, "page_size": page_size})
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,134 @@
|
|||
"""
|
||||
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()
|
||||
|
|
@ -0,0 +1,68 @@
|
|||
"""补充更新 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} 条')
|
||||
|
|
@ -0,0 +1,401 @@
|
|||
"""只采集今天的新帖子 - 优化版"""
|
||||
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,5 +1,7 @@
|
|||
"""一尘网爬虫 - 适配 pm001.net 连体纪念钞板块"""
|
||||
import re
|
||||
import httpx
|
||||
import json
|
||||
from typing import Optional, Dict, List, Any
|
||||
from datetime import datetime, timedelta
|
||||
from bs4 import BeautifulSoup
|
||||
|
|
@ -207,20 +209,78 @@ 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 = [
|
||||
r'(\d+)\s*[万Ww]?\s*元',
|
||||
r'(\d+)\s*-\s*(\d+)\s*[万Ww]?',
|
||||
r'[收售出求兑买]\s*(\d+)',
|
||||
# 第一优先级:明确的价格单位(元、万)
|
||||
patterns_with_unit = [
|
||||
r'(\d+(?:\.\d+)?)\s*[万Ww]?\s*元', # 123元, 1.5万元
|
||||
r'[每单共总]\s*(\d+(?:\.\d+)?)\s*元', # 共123元
|
||||
r'(\d+(?:\.\d+)?)\s*元\s*(?:每|每张|每刀|每条|每套)', # 123元每张
|
||||
]
|
||||
for pattern in patterns:
|
||||
for pattern in patterns_with_unit:
|
||||
m = re.search(pattern, text)
|
||||
if m:
|
||||
try:
|
||||
return float(m.group(1))
|
||||
val = 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]:
|
||||
|
|
@ -237,6 +297,94 @@ 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'):
|
||||
|
|
@ -342,6 +490,8 @@ 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:
|
||||
|
|
|
|||
122
database.py
122
database.py
|
|
@ -1,57 +1,105 @@
|
|||
"""MySQL 数据库连接池模块"""
|
||||
import mysql.connector
|
||||
from mysql.connector import pooling
|
||||
"""PostgreSQL database connection module"""
|
||||
import psycopg2
|
||||
import os
|
||||
from contextlib import contextmanager
|
||||
|
||||
class Database:
|
||||
_pool = None
|
||||
|
||||
@classmethod
|
||||
def get_pool(cls):
|
||||
if cls._pool is None:
|
||||
cls._pool = pooling.MySQLConnectionPool(
|
||||
pool_name="coolbot_pool",
|
||||
pool_size=5,
|
||||
host="127.0.0.1",
|
||||
port=3306,
|
||||
user="root",
|
||||
password="Coolbot123",
|
||||
database="coolbot_data",
|
||||
charset="utf8mb4"
|
||||
)
|
||||
return cls._pool
|
||||
|
||||
@contextmanager
|
||||
def get_connection(self):
|
||||
conn = self.get_pool().get_connection()
|
||||
try:
|
||||
yield conn
|
||||
finally:
|
||||
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"),
|
||||
}
|
||||
|
||||
@contextmanager
|
||||
def get_db():
|
||||
conn = None
|
||||
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):
|
||||
with self.get_connection() as conn:
|
||||
cursor = conn.cursor(dictionary=dictionary)
|
||||
conn = None
|
||||
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:
|
||||
cursor.close()
|
||||
raw_cursor.close()
|
||||
finally:
|
||||
if conn:
|
||||
conn.close()
|
||||
|
||||
db = Database()
|
||||
|
||||
def test_connection():
|
||||
"""测试数据库连接"""
|
||||
try:
|
||||
with db.get_cursor() as cursor:
|
||||
cursor.execute("SELECT 1 as test")
|
||||
result = cursor.fetchone()
|
||||
print(f"✅ 数据库连接成功: {result}")
|
||||
with get_db() as conn:
|
||||
cur = conn.cursor()
|
||||
cur.execute("SELECT version()")
|
||||
print(f"DB OK: {str(cur.fetchone()[0])[:50]}")
|
||||
return True
|
||||
except Exception as e:
|
||||
print(f"❌ 数据库连接失败: {e}")
|
||||
print(f"DB ERROR: {e}")
|
||||
return False
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_connection()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,107 @@
|
|||
#!/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,13 +6,11 @@ uvicorn[standard]>=0.27.0
|
|||
pydantic>=2.5.0
|
||||
|
||||
# Database
|
||||
mysql-connector-python>=8.3.0
|
||||
psycopg2-binary>=2.9.9
|
||||
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
|
||||
|
|
|
|||
|
|
@ -0,0 +1,73 @@
|
|||
"""
|
||||
定时任务调度器 - 修复版
|
||||
集成 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()
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
#!/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) = CURDATE()")
|
||||
cursor.execute("SELECT COUNT(*) as cnt FROM yichens_posts WHERE DATE(crawled_at) = CURRENT_DATE")
|
||||
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) = CURDATE()
|
||||
WHERE price IS NOT NULL AND price > 0 AND DATE(crawled_at) = CURRENT_DATE
|
||||
""")
|
||||
today_price_stats = cursor.fetchone()
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue