From 07b2e5c7b56fd1a3a504d46a47923c41e3b69930 Mon Sep 17 00:00:00 2001 From: caibotmini Date: Mon, 6 Apr 2026 14:41:01 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20v0.0.1=20-=20CoolBotDataSys=20=E5=AE=8C?= =?UTF-8?q?=E6=95=B4=E7=89=88=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - FastAPI REST API (藏品/价格/一尘数据/统计报表) - 爬虫: crawl_today.py (增量) + yichens_spider.py (全量) - PostgreSQL 数据库 + Redis 缓存 - LLM 价格提取模块 - APScheduler 定时调度 - 飞书日报推送 - 修复 CURDATE() -> CURRENT_DATE (PostgreSQL兼容) - 修复 API statistics 端点 - 新增分页数据封装 (PaginatedData) - 新增龙钞分类统计报表 (report/stats) - requirements.txt: mysql->psycopg2-binary --- .env.example | 22 ++ README.md | 120 ++++++++- alembic.ini | 41 ++++ alembic/README | 1 + alembic/env.py | 152 ++++++++++++ alembic/script.py.mako | 28 +++ alembic/versions/9c08601089e0_baseline_v1.py | 28 +++ api/main.py | 2 +- api/middleware/response.py | 8 + api/routes/yichens.py | 243 +++++++++++++++++-- cache.py | 134 ++++++++++ crawlers/backfill_special.py | 68 ++++++ crawlers/crawl_today.py | 49 +++- crawlers/yichens_spider.py | 162 ++++++++++++- database.py | 77 +++++- llm_price_extract.py | 107 ++++++++ requirements.txt | 4 +- scheduler.py | 73 ++++++ scripts/crawl_cron.sh | 19 ++ scripts/daily_report.py | 4 +- 20 files changed, 1295 insertions(+), 47 deletions(-) create mode 100644 .env.example create mode 100644 alembic.ini create mode 100644 alembic/README create mode 100644 alembic/env.py create mode 100644 alembic/script.py.mako create mode 100644 alembic/versions/9c08601089e0_baseline_v1.py create mode 100644 cache.py create mode 100644 crawlers/backfill_special.py create mode 100644 llm_price_extract.py create mode 100644 scheduler.py create mode 100755 scripts/crawl_cron.sh diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..426fac3 --- /dev/null +++ b/.env.example @@ -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 diff --git a/README.md b/README.md index 5f753f2..4e97d60 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,119 @@ -# CoolBotDataSys +# CoolBotDataSys - 酷博特数据分析系统 -酷博特数据分析系统 - 多源数据采集、存储、清洗、分析全链路自动化平台 \ No newline at end of file +**版本：** 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 +``` diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..72747d8 --- /dev/null +++ b/alembic.ini @@ -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 diff --git a/alembic/README b/alembic/README new file mode 100644 index 0000000..98e4f9c --- /dev/null +++ b/alembic/README @@ -0,0 +1 @@ +Generic single-database configuration. \ No newline at end of file diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..e0f9009 --- /dev/null +++ b/alembic/env.py @@ -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() diff --git a/alembic/script.py.mako b/alembic/script.py.mako new file mode 100644 index 0000000..1101630 --- /dev/null +++ b/alembic/script.py.mako @@ -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"} diff --git a/alembic/versions/9c08601089e0_baseline_v1.py b/alembic/versions/9c08601089e0_baseline_v1.py new file mode 100644 index 0000000..7dd525a --- /dev/null +++ b/alembic/versions/9c08601089e0_baseline_v1.py @@ -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 diff --git a/api/main.py b/api/main.py index 39aae5c..d5c55cc 100644 --- a/api/main.py +++ b/api/main.py @@ -106,7 +106,7 @@ async def get_statistics(): # 今日新增价格记录 cursor.execute(""" SELECT COUNT(*) as cnt FROM price_history - WHERE DATE(crawled_at) = CURDATE() + WHERE DATE(crawled_at) = CURRENT_DATE """) today_price_records = cursor.fetchone()["cnt"] diff --git a/api/middleware/response.py b/api/middleware/response.py index 8aa031c..c2a182a 100644 --- a/api/middleware/response.py +++ b/api/middleware/response.py @@ -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, diff --git a/api/routes/yichens.py b/api/routes/yichens.py index fd65bea..d1bea77 100644 --- a/api/routes/yichens.py +++ b/api/routes/yichens.py @@ -4,7 +4,7 @@ from typing import Optional, List from datetime import datetime from api.middleware.response import ApiResponse, PaginatedData -from database import db +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}) + diff --git a/cache.py b/cache.py new file mode 100644 index 0000000..62eef60 --- /dev/null +++ b/cache.py @@ -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() diff --git a/crawlers/backfill_special.py b/crawlers/backfill_special.py new file mode 100644 index 0000000..2b58746 --- /dev/null +++ b/crawlers/backfill_special.py @@ -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} 条') diff --git a/crawlers/crawl_today.py b/crawlers/crawl_today.py index 0ab3da6..92d0de3 100644 --- a/crawlers/crawl_today.py +++ b/crawlers/crawl_today.py @@ -9,6 +9,48 @@ 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，向下到
（签名）为止。 + """ + 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('', 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']*>', '\n', chunk) + text = re.sub(r'', '', text) + text = re.sub(r']*>', '\n', text) + text = re.sub(r'

', '', text) + text = re.sub(r'', '\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__) @@ -105,10 +147,7 @@ class YichensTodaySpider(PaginationSpider): if not price: price, price_unit = self.extract_price(page_text) - content = None - m = re.search(r"(出售|收购|求购|出|售|收)[^\n]{10,500}", page_text) - if m: - content = m.group(0)[:1000] + content = extract_post_content(html) special_types = [] if title: @@ -298,7 +337,7 @@ class YichensTodaySpider(PaginationSpider): post.get("price"), post.get("price_unit"), post.get("special_types"), - None, + post.get("number_features"), post.get("author_username"), f"user_{post.get('post_id')}", post.get("contact"), diff --git a/crawlers/yichens_spider.py b/crawlers/yichens_spider.py index 7b84498..f28e26b 100644 --- a/crawlers/yichens_spider.py +++ b/crawlers/yichens_spider.py @@ -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: diff --git a/database.py b/database.py index 649374e..f0ef7b5 100644 --- a/database.py +++ b/database.py @@ -4,16 +4,15 @@ import os 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'), + "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(): - """Database connection context manager""" conn = None try: conn = psycopg2.connect(**DB_CONFIG) @@ -27,6 +26,70 @@ def get_db(): 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 + 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() + +db = Database() + def test_connection(): try: with get_db() as conn: @@ -38,5 +101,5 @@ def test_connection(): print(f"DB ERROR: {e}") return False -if __name__ == '__main__': +if __name__ == "__main__": test_connection() diff --git a/llm_price_extract.py b/llm_price_extract.py new file mode 100644 index 0000000..9a90423 --- /dev/null +++ b/llm_price_extract.py @@ -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}") diff --git a/requirements.txt b/requirements.txt index 6e7f54c..fac8107 100644 --- a/requirements.txt +++ b/requirements.txt @@ -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 diff --git a/scheduler.py b/scheduler.py new file mode 100644 index 0000000..3b52767 --- /dev/null +++ b/scheduler.py @@ -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() diff --git a/scripts/crawl_cron.sh b/scripts/crawl_cron.sh new file mode 100755 index 0000000..713bbb0 --- /dev/null +++ b/scripts/crawl_cron.sh @@ -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) +" diff --git a/scripts/daily_report.py b/scripts/daily_report.py index c3659b1..c6f1ba4 100644 --- a/scripts/daily_report.py +++ b/scripts/daily_report.py @@ -27,7 +27,7 @@ def get_yichens_stats(): total_posts = cursor.fetchone()["cnt"] # 今日新增 - cursor.execute("SELECT COUNT(*) as cnt FROM yichens_posts WHERE DATE(crawled_at) = 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()