v6ole/GX-gp-notify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
754214692e95c9806db6861b2ef9793a82784cf6
GX-gp-notify/docs/superpowers/plans/2026-05-09-fastapi-docker-migration.md
T

2761 lines
79 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# FastAPI + Docker 迁移实施计划
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** 将广西政府采购网公告监控系统从 Flask CLI 架构迁移到 FastAPI + Docker，含 Spider 模块化、SQLAlchemy ORM、APScheduler 和异步改造。
**Architecture:** 单体 FastAPI 应用，Spider 基类 + Pipeline 策略模式统一爬虫管理，PostgreSQL + asyncpg 异步数据库，APScheduler 内置定时任务，Docker 单容器部署连接外部 PostgreSQL。
**Tech Stack:** Python 3.12, FastAPI, SQLAlchemy 2.0 (async), asyncpg, httpx, APScheduler, pydantic-settings, Alembic, pytest, ruff, Docker, BeautifulSoup4, pycryptodome
---
### Task 1: 创建项目骨架
**Files:**
- Create: `pyproject.toml`
- Create: `app/__init__.py`
- Create: `app/config.py`
- Create: `app/main.py`
- Create: `.env.example`
- Create: `.gitignore`
- [ ] **Step 1: 创建 pyproject.toml**
```toml
[project]
name = "gx-gp-notify"
version = "2.0.0"
description = "广西政府采购网公告监控系统"
requires-python = ">=3.12"
dependencies = [
"fastapi>=0.115.0",
"uvicorn[standard]>=0.30.0",
"sqlalchemy[asyncio]>=2.0.30",
"asyncpg>=0.29.0",
"alembic>=1.13.0",
"httpx>=0.27.0",
"apscheduler>=3.10.0",
"pydantic-settings>=2.3.0",
"beautifulsoup4>=4.12.0",
"lxml>=5.2.0",
"pycryptodome>=3.20.0",
"python-dateutil>=2.9.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.2.0",
"pytest-asyncio>=0.23.0",
"pytest-cov>=5.0.0",
"httpx>=0.27.0",
"ruff>=0.4.0",
]
[tool.ruff]
target-version = "py312"
line-length = 100
[tool.ruff.lint]
select = ["E", "F", "I", "N", "W", "UP"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]
```
- [ ] **Step 2: 创建 .env.example**
```bash
# 应用
DEBUG=false
LOG_LEVEL=INFO
# 数据库
DATABASE_URL=postgresql+asyncpg://gx-gp-notify:password@10.10.10.14:5432/gx-gp-notify
# 爬虫
CRAWLER_BASE_URL=https://zfcg.gxzf.gov.cn
CRAWLER_KEYWORDS=["大化"]
CRAWLER_MAX_PAGES=10
CRAWLER_TIMEOUT=30
# 企业微信
WECHAT_ENABLED=true
WECHAT_CORP_ID=ww69e8e44636f47780
WECHAT_AGENT_ID=1000007
WECHAT_SECRET=
WECHAT_TOKEN=
WECHAT_ENCODING_AES_KEY=
WECHAT_PORT=18001
WECHAT_HOST=0.0.0.0
# 定时任务
SCHEDULER_ENABLED=true
SCHEDULER_CRON=0 8,14,18 * * *
# Markdown
MARKDOWN_ENABLED=true
MARKDOWN_OUTPUT_FILE=onu.md
```
- [ ] **Step 3: 创建 .gitignore**
```gitignore
.env
logs/
*.log
__pycache__/
*.pyc
.venv/
.ruff_cache/
.pytest_cache/
*.egg-info/
dist/
```
- [ ] **Step 4: 创建 app/__init__.py** (空文件)
- [ ] **Step 5: 创建 app/config.py**
```python
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import List
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
# 应用
debug: bool = False
log_level: str = "INFO"
# 数据库
database_url: str = "postgresql+asyncpg://gx-gp-notify:password@localhost:5432/gx-gp-notify"
# 爬虫
crawler_base_url: str = "https://zfcg.gxzf.gov.cn"
crawler_keywords: List[str] = ["大化"]
crawler_max_pages: int = 10
crawler_timeout: int = 30
crawler_page_size: int = 100
# 企业微信
wechat_enabled: bool = True
wechat_corp_id: str = ""
wechat_agent_id: str = ""
wechat_secret: str = ""
wechat_token: str = ""
wechat_encoding_aes_key: str = ""
wechat_port: int = 18001
wechat_host: str = "0.0.0.0"
# 定时任务
scheduler_enabled: bool = True
scheduler_cron: str = "0 8,14,18 * * *"
# Markdown
markdown_enabled: bool = True
markdown_output_file: str = "onu.md"
# 公告来源（JSON 字符串，从环境变量读取）
announcement_sources: str = '{"ZcyAnnouncement1":{"category_id":66485,"name":"采购公告","type":"purchase"},"ZcyAnnouncement2":{"category_id":66485,"name":"结果公告","type":"result"},"ZcyAnnouncement3":{"category_id":66485,"name":"合同公告","type":"contract"},"ZcyAnnouncement4":{"category_id":66485,"name":"更正公告","type":"correction"},"ZcyAnnouncement5":{"category_id":66485,"name":"招标文件预公示","type":"pre_announcement"},"ZcyAnnouncement6":{"category_id":66485,"name":"单一来源公示","type":"single_source"},"ZcyAnnouncement7":{"category_id":66485,"name":"电子卖场公示","type":"electronic_market"},"ZcyAnnouncement10":{"category_id":66485,"name":"履约验收公示","type":"acceptance"},"ZcyAnnouncement11":{"category_id":66485,"name":"工程类公告","type":"engineering"},"ZcyAnnouncement20":{"category_id":66485,"name":"框架协议征集公告","type":"framework_agreement"},"ZcyAnnouncement21":{"category_id":66485,"name":"框架协议入围结果公告","type":"framework_result"},"ZcyAnnouncement23":{"category_id":66485,"name":"框架协议成交结果汇总公告","type":"framework_summary"},"61-266648":{"category_id":66485,"name":"采购意向公开","type":"intention"}}'
settings = Settings()
```
- [ ] **Step 6: 创建 app/main.py**
```python
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.config import settings
@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动时初始化
logging.basicConfig(level=settings.log_level)
yield
# 关闭时清理
app = FastAPI(
title="广西政府采购网公告监控系统",
version="2.0.0",
lifespan=lifespan,
docs_url="/docs" if settings.debug else None,
redoc_url=None,
)
@app.get("/health")
async def health():
return {"status": "ok"}
```
- [ ] **Step 7: 安装依赖并验证**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
pip install -e ".[dev]"
uvicorn app.main:app --host 0.0.0.0 --port 8000 &
sleep 2
curl http://localhost:8000/health
# Expected: {"status":"ok"}
kill %1
```
- [ ] **Step 8: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add pyproject.toml app/__init__.py app/config.py app/main.py .env.example .gitignore
git commit -m "feat: 创建 FastAPI 项目骨架（config + main + 依赖管理）"
```
---
### Task 2: 数据库模型 + Alembic
**Files:**
- Create: `app/models/__init__.py`
- Create: `app/models/announcement.py`
- Create: `app/models/schemas.py`
- Create: `alembic.ini`
- Create: `alembic/env.py`
- Create: `alembic/script.py.mako`
- Modify: `app/main.py` — 添加数据库引擎
- [ ] **Step 1: 创建 app/models/__init__.py** (空文件)
- [ ] **Step 2: 创建 app/models/announcement.py**
```python
import hashlib
from datetime import datetime, date
from sqlalchemy import String, Boolean, DateTime, Integer, Text, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase):
pass
class Announcement(Base):
__tablename__ = "announcements"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
title: Mapped[str] = mapped_column(String(500), nullable=False)
publish_date: Mapped[datetime] = mapped_column(DateTime, nullable=False)
purchase_name: Mapped[str] = mapped_column(String(200), default="")
content_url: Mapped[str] = mapped_column(Text, default="")
source_code: Mapped[str] = mapped_column(String(50), nullable=False)
source_name: Mapped[str] = mapped_column(String(100), nullable=False)
announcement_type: Mapped[str] = mapped_column(String(50), default="purchase")
content_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
crawl_mode: Mapped[str] = mapped_column(String(20), default="auto")
is_new: Mapped[bool] = mapped_column(Boolean, default=True)
is_sent: Mapped[bool] = mapped_column(Boolean, default=False)
keyword_matched: Mapped[bool] = mapped_column(Boolean, default=False)
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())
@staticmethod
def generate_hash(title: str, publish_date: str, purchase_name: str,
content_url: str, source_code: str) -> str:
content = f"{title}|{publish_date}|{purchase_name}|{content_url}|{source_code}"
return hashlib.sha256(content.encode("utf-8")).hexdigest()
@staticmethod
def source_map() -> dict:
import json
from app.config import settings
return json.loads(settings.announcement_sources)
```
- [ ] **Step 3: 创建 app/models/schemas.py**
```python
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
class AnnouncementResponse(BaseModel):
id: int
title: str
publish_date: datetime
purchase_name: str
content_url: str
source_code: str
source_name: str
announcement_type: str
crawl_mode: str
is_new: bool
is_sent: bool
keyword_matched: bool
created_at: datetime
model_config = {"from_attributes": True}
class AnnouncementListResponse(BaseModel):
total: int
page: int
page_size: int
items: list[AnnouncementResponse]
class CrawlTriggerRequest(BaseModel):
keywords: Optional[list[str]] = None
sources: Optional[list[str]] = None
manual: bool = False
class CrawlStatusResponse(BaseModel):
running: bool
last_crawl_time: Optional[datetime] = None
total_sources: int
class SourceInfo(BaseModel):
code: str
name: str
type: str
class SourcesResponse(BaseModel):
sources: list[SourceInfo]
class JobResponse(BaseModel):
id: str
name: str
next_run_time: Optional[str] = None
```
- [ ] **Step 4: 初始化 Alembic**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
pip install alembic
alembic init alembic
```
- [ ] **Step 5: 修改 alembic/env.py**
Read the generated file, replace its content with:
```python
from alembic import context
from sqlalchemy import engine_from_config, pool
from app.models.announcement import Base
config = context.config
config.set_main_option("sqlalchemy.url", "postgresql+asyncpg://placeholder:placeholder@localhost:5432/placeholder")
target_metadata = Base.metadata
def run_migrations_offline():
url = config.get_main_option("sqlalchemy.url")
context.configure(url=url, target_metadata=target_metadata, literal_binds=True)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online():
from app.config import settings
connectable = engine_from_config(
{"sqlalchemy.url": settings.database_url},
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()
```
- [ ] **Step 6: 生成初始迁移**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
DATABASE_URL=postgresql+asyncpg://gx-gp-notify:MA6RBX4F6Bd5DGmw@10.10.10.14:5432/gx-gp-notify \
alembic revision --autogenerate -m "create_announcements_table"
# Expected: Generating .../alembic/versions/xxxx_create_announcements_table.py ... done
```
- [ ] **Step 7: 运行迁移**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
DATABASE_URL=postgresql+asyncpg://gx-gp-notify:MA6RBX4F6Bd5DGmw@10.10.10.14:5432/gx-gp-notify \
alembic upgrade head
# Expected: Running upgrade ... -> xxxx, create announcements table
```
- [ ] **Step 8: 更新 app/main.py 添加数据库初始化**
```python
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.config import settings
from app.models.announcement import Base
engine = create_async_engine(settings.database_url, echo=settings.debug)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def get_db() -> AsyncSession:
async with async_session() as session:
yield session
@asynccontextmanager
async def lifespan(app: FastAPI):
logging.basicConfig(
level=getattr(logging, settings.log_level),
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
yield
await engine.dispose()
app = FastAPI(
title="广西政府采购网公告监控系统",
version="2.0.0",
lifespan=lifespan,
docs_url="/docs" if settings.debug else None,
redoc_url=None,
)
@app.get("/health")
async def health():
return {"status": "ok"}
```
- [ ] **Step 9: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/models/ alembic/ alembic.ini app/main.py
git commit -m "feat: 添加 SQLAlchemy 模型 + Alembic 数据库迁移"
```
---
### Task 3: Spider 基类 + Pipeline 配置
**Files:**
- Create: `app/crawler/__init__.py`
- Create: `app/crawler/base.py`
- Test: `tests/test_crawler/test_base.py`
- Create: `tests/__init__.py`
- Create: `tests/test_crawler/__init__.py`
- Create: `tests/conftest.py`
- [ ] **Step 1: 创建 app/crawler/__init__.py** (空文件)
- [ ] **Step 2: 创建 app/crawler/base.py**
```python
import hashlib
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional
@dataclass
class CrawlResult:
source_code: str
source_name: str
total_count: int = 0
new_count: int = 0
announcements: list = field(default_factory=list)
error_message: Optional[str] = None
crawled_at: datetime = field(default_factory=datetime.now)
duration: float = 0.0
@property
def success(self) -> bool:
return self.error_message is None
@dataclass
class PipelineConfig:
filter_enabled: bool = True
keywords: List[str] = field(default_factory=list)
dedup_enabled: bool = True
notify_mode: str = "filtered"
mark_sent: bool = False
@dataclass
class PipelineResult:
stored: int = 0
filtered: int = 0
notified: int = 0
markdown_generated: bool = False
class BaseSpider(ABC):
name: str
source_code: str
source_name: str
@abstractmethod
async def crawl(self) -> CrawlResult:
...
def get_pipeline_config(self) -> PipelineConfig:
return PipelineConfig()
@staticmethod
def generate_content_hash(title: str, publish_date: str, purchase_name: str,
content_url: str, source_code: str) -> str:
content = f"{title}|{publish_date}|{purchase_name}|{content_url}|{source_code}"
return hashlib.sha256(content.encode("utf-8")).hexdigest()
```
- [ ] **Step 3: 创建 tests/__init__.py, tests/test_crawler/__init__.py** (空文件)
- [ ] **Step 4: 创建 tests/conftest.py**
```python
import pytest
@pytest.fixture
def sample_announcement_data():
return {
"title": "测试公告标题",
"publish_date": "2026-05-09",
"purchase_name": "测试单位",
"content_url": "https://example.com/detail/123",
"source_code": "ZcyAnnouncement1",
"source_name": "采购公告",
}
```
- [ ] **Step 5: 创建 tests/test_crawler/test_base.py**
```python
import pytest
from datetime import datetime
from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig, PipelineResult
class FakeSpider(BaseSpider):
name = "test_spider"
source_code = "test_source"
source_name = "测试来源"
async def crawl(self) -> CrawlResult:
return CrawlResult(
source_code=self.source_code,
source_name=self.source_name,
total_count=5,
new_count=3,
crawled_at=datetime.now(),
)
def get_pipeline_config(self) -> PipelineConfig:
return PipelineConfig(
filter_enabled=True,
keywords=["测试"],
dedup_enabled=True,
notify_mode="filtered",
mark_sent=False,
)
@pytest.mark.asyncio
async def test_base_spider_crawl():
spider = FakeSpider()
result = await spider.crawl()
assert result.source_code == "test_source"
assert result.total_count == 5
assert result.new_count == 3
assert result.success is True
def test_pipeline_config_defaults():
config = PipelineConfig()
assert config.filter_enabled is True
assert config.keywords == []
assert config.dedup_enabled is True
assert config.notify_mode == "filtered"
assert config.mark_sent is False
def test_pipeline_result_defaults():
result = PipelineResult()
assert result.stored == 0
assert result.filtered == 0
assert result.notified == 0
assert result.markdown_generated is False
def test_generate_content_hash():
h = BaseSpider.generate_content_hash(
"title", "2026-05-09", "unit", "https://x.com", "source"
)
assert len(h) == 64
assert all(c in "0123456789abcdef" for c in h)
def test_generate_content_hash_deterministic():
h1 = BaseSpider.generate_content_hash("t", "d", "p", "u", "s")
h2 = BaseSpider.generate_content_hash("t", "d", "p", "u", "s")
assert h1 == h2
def test_crawl_result_failure():
result = CrawlResult(
source_code="s", source_name="n",
error_message="连接超时"
)
assert result.success is False
```
- [ ] **Step 6: 运行测试**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_crawler/test_base.py -v
# Expected: 6 passed
```
- [ ] **Step 7: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/crawler/ tests/
git commit -m "feat: 添加 Spider 基类 + Pipeline 配置 + 测试"
```
---
### Task 4: Parssers 模块
**Files:**
- Create: `app/crawler/parsers.py`
- Test: `tests/test_crawler/test_parsers.py`
- [ ] **Step 1: 创建测试 tests/test_crawler/test_parsers.py**
```python
import json
from datetime import datetime
from app.crawler.parsers import (
parse_gxgp_api_response,
parse_dahuagov_html,
extract_pagination,
)
def make_api_response(records_data):
return {
"success": True,
"result": {
"data": {
"data": records_data,
"total": len(records_data),
"pageNo": 1,
"pageSize": 100,
"pages": 1,
"empty": len(records_data) == 0,
"hasNext": False,
"hasPrevious": False,
}
},
}
def test_parse_gxgp_single_record():
response = make_api_response([
{
"title": "测试采购公告",
"publishDate": 1746720000000,
"purchaseName": "测试采购单位",
"articleId": 12345,
}
])
crawled_at = datetime(2026, 5, 9, 10, 0, 0)
results = parse_gxgp_api_response(
response,
source_code="ZcyAnnouncement1",
source_name="采购公告",
crawled_at=crawled_at,
category_id=66485,
)
assert len(results) == 1
assert results[0]["title"] == "测试采购公告"
assert results[0]["source_code"] == "ZcyAnnouncement1"
assert "content_hash" in results[0]
def test_parse_gxgp_empty_response():
response = make_api_response([])
crawled_at = datetime(2026, 5, 9, 10, 0, 0)
results = parse_gxgp_api_response(
response, "ZcyAnnouncement1", "采购公告", crawled_at, 66485
)
assert len(results) == 0
def test_parse_gxgp_missing_title():
response = make_api_response([
{"title": "", "publishDate": 1746720000000, "purchaseName": "x", "articleId": 1}
])
crawled_at = datetime(2026, 5, 9, 10, 0, 0)
results = parse_gxgp_api_response(
response, "ZcyAnnouncement1", "采购公告", crawled_at, 66485
)
assert len(results) == 0
def test_extract_pagination():
response = make_api_response([])
pagination = extract_pagination(response)
assert pagination["total"] == 0
assert pagination["page_no"] == 1
assert pagination["has_next"] is False
def test_parse_dahuagov_html():
html = """
<html><body>
<ul class="more-list">
<li>
<span>2026-05-08</span>
<a href="./detail/123.html" title="大化县某项目采购公告">大化县某项目采购公告</a>
</li>
<li>
<span>2026-05-07</span>
<a href="./detail/124.html" title="大化县另一采购公告">大化县另一采购公告</a>
</li>
</ul>
</body></html>
"""
crawled_at = datetime(2026, 5, 9, 10, 0, 0)
results = parse_dahuagov_html(html, crawled_at)
assert len(results) == 2
assert results[0]["title"] == "大化县某项目采购公告"
assert results[0]["source_code"] == "dahuagov"
assert results[0]["source_name"] == "大化县政府网采购公告"
assert results[0]["purchase_name"] == "大化瑶族自治县"
def test_parse_dahuagov_html_no_list():
html = "<html><body></body></html>"
crawled_at = datetime(2026, 5, 9, 10, 0, 0)
results = parse_dahuagov_html(html, crawled_at)
assert len(results) == 0
```
- [ ] **Step 2: 运行测试确认失败**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_crawler/test_parsers.py -v
# Expected: FAIL — ModuleNotFoundError or ImportError
```
- [ ] **Step 3: 创建 app/crawler/parsers.py**
```python
import hashlib
from datetime import datetime
from typing import Any, Dict, List, Optional
from urllib.parse import urljoin
from bs4 import BeautifulSoup
def parse_gxgp_api_response(
response_data: Dict[str, Any],
source_code: str,
source_name: str,
crawled_at: datetime,
category_id: int,
) -> List[Dict[str, Any]]:
if not response_data.get("success"):
return []
data = response_data.get("result", {}).get("data", {})
records = data.get("data", [])
if not records:
return []
results = []
for record in records:
title = str(record.get("title", "")).strip()
if not title:
continue
timestamp = record.get("publishDate")
if not timestamp:
continue
try:
publish_date = datetime.fromtimestamp(int(timestamp) / 1000)
except (ValueError, TypeError):
continue
purchase_name = str(record.get("purchaseName", "")).strip()
article_id = record.get("articleId")
if not article_id:
continue
content_url = f"https://zfcg.gxzf.gov.cn/site/detail?parentId={category_id}&articleId={article_id}"
announce = {
"title": title,
"publish_date": publish_date,
"purchase_name": purchase_name,
"content_url": content_url,
"source_code": source_code,
"source_name": source_name,
"announcement_type": "purchase",
"crawl_mode": "auto",
"is_new": True,
"is_today": publish_date.date() == datetime.now().date(),
}
announce["content_hash"] = _generate_hash(announce)
results.append(announce)
return results
def extract_pagination(response_data: Dict[str, Any]) -> Dict[str, Any]:
data = response_data.get("result", {}).get("data", {})
return {
"total": data.get("total", 0),
"page_no": data.get("pageNo", 1),
"page_size": data.get("pageSize", 100),
"pages": data.get("pages", 0),
"empty": data.get("empty", True),
"has_next": data.get("hasNext", False),
"has_previous": data.get("hasPrevious", False),
}
def parse_dahuagov_html(html: str, crawled_at: datetime) -> List[Dict[str, Any]]:
soup = BeautifulSoup(html, "html.parser")
lists = soup.find_all("ul", class_="more-list")
if not lists:
return []
results = []
base_url = "http://www.gxdh.gov.cn"
base_path = "/xxgk/zdlyxxgk/ggzypzly/zfcgly/cggg/"
for ul in lists:
for li in ul.find_all("li"):
date_span = li.find("span")
if not date_span:
continue
date_text = date_span.get_text(strip=True)
try:
publish_date = datetime.strptime(date_text, "%Y-%m-%d")
except ValueError:
continue
link_tag = li.find("a")
if not link_tag:
continue
title = link_tag.get("title", "") or link_tag.get_text(strip=True)
href = link_tag.get("href", "")
if not title or not href:
continue
if href.startswith("./") or href.startswith("../"):
content_url = urljoin(base_url + base_path, href)
elif href.startswith("/"):
content_url = base_url + href
elif href.startswith("http"):
content_url = href
else:
content_url = urljoin(base_url + base_path, href)
announce = {
"title": title,
"publish_date": publish_date,
"purchase_name": "大化瑶族自治县",
"content_url": content_url,
"source_code": "dahuagov",
"source_name": "大化县政府网采购公告",
"announcement_type": "purchase",
"crawl_mode": "auto",
"is_new": True,
"is_today": publish_date.date() == datetime.now().date(),
}
announce["content_hash"] = _generate_hash(announce)
results.append(announce)
return results
def _generate_hash(ann: Dict[str, Any]) -> str:
content = (
f"{ann['title']}|{ann['publish_date'].strftime('%Y-%m-%d')}"
f"|{ann['purchase_name']}|{ann['content_url']}|{ann['source_code']}"
)
return hashlib.sha256(content.encode("utf-8")).hexdigest()
```
- [ ] **Step 4: 运行测试验证通过**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_crawler/test_parsers.py -v
# Expected: 6 passed
```
- [ ] **Step 5: 运行全部测试**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/ -v
# Expected: 12 passed
```
- [ ] **Step 6: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/crawler/parsers.py tests/test_crawler/test_parsers.py
git commit -m "feat: 添加数据解析器模块 + 测试"
```
---
### Task 5: GXGP Spider（广西政府采购网）
**Files:**
- Create: `app/crawler/gxgp_spider.py`
- Test: `tests/test_crawler/test_gxgp_spider.py`
- [ ] **Step 1: 创建测试 tests/test_crawler/test_gxgp_spider.py**
```python
import pytest
from datetime import datetime
from unittest.mock import AsyncMock, patch, MagicMock
from app.crawler.gxgp_spider import GXGPSpider
from app.crawler.base import PipelineConfig
@pytest.mark.asyncio
async def test_gxgp_spider_attributes():
spider = GXGPSpider()
assert spider.name == "gxgp"
assert spider.source_code == "gxgp"
assert spider.source_name == "广西政府采购网"
@pytest.mark.asyncio
async def test_gxgp_spider_pipeline_config():
spider = GXGPSpider()
config = spider.get_pipeline_config()
assert isinstance(config, PipelineConfig)
assert config.filter_enabled is True
assert config.notify_mode == "filtered"
assert config.mark_sent is False
@pytest.mark.asyncio
async def test_gxgp_spider_crawl_empty():
spider = GXGPSpider()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = {
"success": True,
"result": {"data": {"data": [], "total": 0, "pageNo": 1, "pageSize": 100,
"pages": 0, "empty": True, "hasNext": False, "hasPrevious": False}},
}
with patch("httpx.AsyncClient.post", AsyncMock(return_value=mock_response)):
result = await spider.crawl()
assert result.total_count == 0
assert len(result.announcements) == 0
assert result.success is True
@pytest.mark.asyncio
async def test_gxgp_spider_crawl_with_data():
spider = GXGPSpider()
mock_data = {
"success": True,
"result": {
"data": {
"data": [{
"title": "测试采购公告",
"publishDate": 1746720000000,
"purchaseName": "测试单位",
"articleId": 12345,
}],
"total": 1, "pageNo": 1, "pageSize": 100,
"pages": 1, "empty": False, "hasNext": False, "hasPrevious": False,
}
},
}
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.json.return_value = mock_data
with patch("httpx.AsyncClient.post", AsyncMock(return_value=mock_response)):
result = await spider.crawl(sources=["ZcyAnnouncement1"])
assert result.total_count >= 0
assert result.success is True
```
- [ ] **Step 2: 运行测试确认失败**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_crawler/test_gxgp_spider.py -v
# Expected: FAIL — ImportError
```
- [ ] **Step 3: 创建 app/crawler/gxgp_spider.py**
```python
import json
import random
import time
from datetime import datetime
from typing import List, Optional
import httpx
from app.config import settings
from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig
from app.crawler.parsers import parse_gxgp_api_response, extract_pagination
class GXGPSpider(BaseSpider):
name = "gxgp"
source_code = "gxgp"
source_name = "广西政府采购网"
def __init__(self):
self.base_url = settings.crawler_base_url
self.announcement_api = f"{self.base_url}/portal/category"
def get_pipeline_config(self) -> PipelineConfig:
return PipelineConfig(
filter_enabled=True,
keywords=list(settings.crawler_keywords),
dedup_enabled=True,
notify_mode="filtered",
mark_sent=False,
)
async def crawl(self, sources: Optional[List[str]] = None,
max_pages: Optional[int] = None) -> CrawlResult:
if max_pages is None:
max_pages = settings.crawler_max_pages
if sources is None:
source_map = json.loads(settings.announcement_sources)
sources = list(source_map.keys())
start_time = datetime.now()
all_announcements = []
error_messages = []
async with httpx.AsyncClient(timeout=settings.crawler_timeout) as client:
for source_code in sources:
source_info = json.loads(settings.announcement_sources).get(source_code)
if not source_info:
continue
category_id = source_info["category_id"]
source_name = source_info["name"]
for page_no in range(1, max_pages + 1):
if page_no > 1:
await self._delay()
try:
data = await self._fetch_page(
client, source_code, category_id, page_no
)
if data is None:
break
records = parse_gxgp_api_response(
data, source_code, source_name,
start_time, category_id
)
if not records:
break
all_announcements.extend(records)
pagination = extract_pagination(data)
if not pagination["has_next"]:
break
except Exception as e:
error_messages.append(f"{source_code} page {page_no}: {e}")
break
duration = (datetime.now() - start_time).total_seconds()
return CrawlResult(
source_code=self.source_code,
source_name=self.source_name,
total_count=len(all_announcements),
new_count=len(all_announcements),
announcements=all_announcements,
error_message="; ".join(error_messages) if error_messages else None,
crawled_at=start_time,
duration=duration,
)
async def _fetch_page(self, client: httpx.AsyncClient, source_code: str,
category_id: int, page_no: int) -> Optional[dict]:
payload = {
"keyword": "",
"publishDateBegin": "",
"publishDateEnd": "",
"pageNo": page_no,
"pageSize": settings.crawler_page_size,
"categoryCode": source_code,
"_t": int(time.time() * 1000),
}
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Content-Type": "application/json;charset=UTF-8",
"Origin": self.base_url,
"Referer": f"{self.base_url}/site/category?parentId={category_id}&childrenCode={source_code}",
}
response = await client.post(
self.announcement_api, json=payload, headers=headers
)
if response.status_code != 200:
return None
return response.json()
async def _delay(self):
import asyncio
delay = random.uniform(1.0, 3.0)
await asyncio.sleep(delay)
```
- [ ] **Step 4: 运行测试验证通过**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_crawler/test_gxgp_spider.py -v
# Expected: 3 passed
```
- [ ] **Step 5: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/crawler/gxgp_spider.py tests/test_crawler/test_gxgp_spider.py
git commit -m "feat: 添加 GXGP Spider（广西政府采购网爬虫）+ 测试"
```
---
### Task 6: Dahuagov Spider（大化县政府网）
**Files:**
- Create: `app/crawler/dahuagov_spider.py`
- Test: `tests/test_crawler/test_dahuagov_spider.py`
- [ ] **Step 1: 创建测试 tests/test_crawler/test_dahuagov_spider.py**
```python
import pytest
from unittest.mock import AsyncMock, patch, MagicMock
from app.crawler.dahuagov_spider import DahuagovSpider
from app.crawler.base import PipelineConfig
@pytest.mark.asyncio
async def test_dahuagov_spider_attributes():
spider = DahuagovSpider()
assert spider.name == "dahuagov"
assert spider.source_code == "dahuagov"
assert spider.source_name == "大化县政府网采购公告"
@pytest.mark.asyncio
async def test_dahuagov_spider_pipeline_config():
spider = DahuagovSpider()
config = spider.get_pipeline_config()
assert isinstance(config, PipelineConfig)
assert config.filter_enabled is False
assert config.notify_mode == "all"
assert config.mark_sent is True
@pytest.mark.asyncio
async def test_dahuagov_spider_crawl():
spider = DahuagovSpider()
html = """
<html><body>
<ul class="more-list">
<li><span>2026-05-08</span>
<a href="./detail/1.html" title="公告1">公告1</a></li>
<li><span>2026-05-07</span>
<a href="./detail/2.html" title="公告2">公告2</a></li>
</ul>
</body></html>
"""
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.text = html
with patch("httpx.AsyncClient.get", AsyncMock(return_value=mock_response)):
result = await spider.crawl()
assert result.total_count == 2
assert len(result.announcements) == 2
assert result.success is True
@pytest.mark.asyncio
async def test_dahuagov_spider_crawl_empty():
spider = DahuagovSpider()
mock_response = MagicMock()
mock_response.status_code = 200
mock_response.text = "<html><body></body></html>"
with patch("httpx.AsyncClient.get", AsyncMock(return_value=mock_response)):
result = await spider.crawl()
assert result.total_count == 0
assert result.success is True
```
- [ ] **Step 2: 运行测试确认失败**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_crawler/test_dahuagov_spider.py -v
# Expected: FAIL — ImportError
```
- [ ] **Step 3: 创建 app/crawler/dahuagov_spider.py**
```python
import asyncio
import random
from datetime import datetime
import httpx
from app.config import settings
from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig
from app.crawler.parsers import parse_dahuagov_html
class DahuagovSpider(BaseSpider):
name = "dahuagov"
source_code = "dahuagov"
source_name = "大化县政府网采购公告"
BASE_URL = "http://www.gxdh.gov.cn"
ANNOUNCEMENT_PATH = "/xxgk/zdlyxxgk/ggzypzly/zfcgly/cggg/"
def get_pipeline_config(self) -> PipelineConfig:
return PipelineConfig(
filter_enabled=False,
keywords=[],
dedup_enabled=True,
notify_mode="all",
mark_sent=True,
)
async def crawl(self) -> CrawlResult:
start_time = datetime.now()
url = self.BASE_URL + self.ANNOUNCEMENT_PATH
async with httpx.AsyncClient(timeout=settings.crawler_timeout) as client:
await self._delay()
try:
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "text/html,application/xhtml+xml",
"Accept-Language": "zh-CN,zh;q=0.9",
"Referer": self.BASE_URL,
}
response = await client.get(url, headers=headers)
if response.status_code != 200:
return CrawlResult(
source_code=self.source_code,
source_name=self.source_name,
error_message=f"HTTP {response.status_code}",
crawled_at=start_time,
)
html = response.text
except Exception as e:
return CrawlResult(
source_code=self.source_code,
source_name=self.source_name,
error_message=str(e),
crawled_at=start_time,
)
announcements = parse_dahuagov_html(html, start_time)
duration = (datetime.now() - start_time).total_seconds()
return CrawlResult(
source_code=self.source_code,
source_name=self.source_name,
total_count=len(announcements),
new_count=len(announcements),
announcements=announcements,
crawled_at=start_time,
duration=duration,
)
async def _delay(self):
delay = random.uniform(1.0, 3.0)
await asyncio.sleep(delay)
```
- [ ] **Step 4: 运行测试验证通过**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_crawler/test_dahuagov_spider.py -v
# Expected: 4 passed
```
- [ ] **Step 5: 运行全部测试**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/ -v
# Expected: 19 passed
```
- [ ] **Step 6: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/crawler/dahuagov_spider.py tests/test_crawler/test_dahuagov_spider.py
git commit -m "feat: 添加 Dahuagov Spider（大化县政府网爬虫）+ 测试"
```
---
### Task 7: CrawlService（爬取编排器）
**Files:**
- Create: `app/services/__init__.py`
- Create: `app/services/crawl_service.py`
- Test: `tests/test_services/__init__.py`
- Test: `tests/test_services/test_crawl_service.py`
- [ ] **Step 1: 创建测试 tests/test_services/test_crawl_service.py**
```python
import pytest
from datetime import datetime
from unittest.mock import AsyncMock, patch
from app.crawler.base import BaseSpider, CrawlResult, PipelineConfig
from app.services.crawl_service import CrawlService
class MockGXGPSpider(BaseSpider):
name = "mock_gxgp"
source_code = "mock_gxgp"
source_name = "Mock GXGP"
async def crawl(self, **kwargs):
return CrawlResult(
source_code=self.source_code,
source_name=self.source_name,
total_count=10,
new_count=5,
announcements=[
{
"title": f"公告{i}",
"publish_date": datetime(2026, 5, 9),
"purchase_name": "单位",
"content_url": f"https://x.com/{i}",
"source_code": "ZcyAnnouncement1",
"source_name": "采购公告",
"announcement_type": "purchase",
"crawl_mode": "auto",
"is_new": True,
"is_today": True,
"content_hash": f"hash{i}",
}
for i in range(10)
],
)
def get_pipeline_config(self):
return PipelineConfig(
filter_enabled=True,
keywords=["大化"],
notify_mode="filtered",
)
class MockDahuagovSpider(BaseSpider):
name = "mock_dahuagov"
source_code = "mock_dahuagov"
source_name = "Mock Dahuagov"
async def crawl(self, **kwargs):
return CrawlResult(
source_code=self.source_code,
source_name=self.source_name,
total_count=3,
new_count=3,
announcements=[
{
"title": f"大化公告{j}",
"publish_date": datetime(2026, 5, 9),
"purchase_name": "大化瑶族自治县",
"content_url": f"https://dh.com/{j}",
"source_code": "dahuagov",
"source_name": "大化县政府网采购公告",
"announcement_type": "purchase",
"crawl_mode": "auto",
"is_new": True,
"is_today": True,
"content_hash": f"dh_hash{j}",
}
for j in range(3)
],
)
def get_pipeline_config(self):
return PipelineConfig(
filter_enabled=False,
notify_mode="all",
mark_sent=True,
)
@pytest.mark.asyncio
async def test_crawl_service_registers_spiders():
service = CrawlService()
service.register(MockGXGPSpider())
service.register(MockDahuagovSpider())
assert len(service.spiders) == 2
@pytest.mark.asyncio
async def test_crawl_service_run_all():
service = CrawlService()
service.register(MockGXGPSpider())
service.register(MockDahuagovSpider())
results = await service.run_all()
assert len(results) == 2
assert results[0].total_count == 10
assert results[1].total_count == 3
assert all(r.success for r in results)
@pytest.mark.asyncio
async def test_crawl_service_run_specific():
service = CrawlService()
service.register(MockGXGPSpider())
service.register(MockDahuagovSpider())
results = await service.run_spider("mock_dahuagov")
assert len(results) == 1
assert results[0].source_code == "mock_dahuagov"
```
- [ ] **Step 2: 运行测试确认失败**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_services/test_crawl_service.py -v
# Expected: FAIL — ImportError
```
- [ ] **Step 3: 创建 app/services/__init__.py** (空文件)
- [ ] **Step 4: 创建 app/services/crawl_service.py**
```python
from typing import Dict, List
from app.crawler.base import BaseSpider, CrawlResult
class CrawlService:
def __init__(self):
self.spiders: Dict[str, BaseSpider] = {}
def register(self, spider: BaseSpider):
self.spiders[spider.name] = spider
async def run_all(self) -> List[CrawlResult]:
results = []
for name, spider in self.spiders.items():
result = await spider.crawl()
results.append(result)
return results
async def run_spider(self, name: str, **kwargs) -> List[CrawlResult]:
spider = self.spiders.get(name)
if spider is None:
return [CrawlResult(
source_code=name, source_name=name,
error_message=f"Spider not found: {name}"
)]
result = await spider.crawl(**kwargs)
return [result]
def get_spider_names(self) -> List[str]:
return list(self.spiders.keys())
def get_pipeline_config(self, name: str):
spider = self.spiders.get(name)
if spider:
return spider.get_pipeline_config()
return None
```
- [ ] **Step 5: 运行测试验证通过**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_services/test_crawl_service.py -v
# Expected: 3 passed
```
- [ ] **Step 6: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/services/ tests/test_services/
git commit -m "feat: 添加 CrawlService 爬取编排器 + 测试"
```
---
### Task 8: PostCrawlPipeline 统一管道
**Files:**
- Create: `app/services/pipeline.py`
- Create: `app/services/filter_service.py`
- Test: `tests/test_services/test_pipeline.py`
- Test: `tests/test_services/test_filter_service.py`
- [ ] **Step 1: 创建过滤测试 tests/test_services/test_filter_service.py**
```python
from datetime import datetime
from app.services.filter_service import filter_by_keywords, filter_by_date
def test_filter_by_keywords_match():
announcements = [
{"title": "大化县采购公告", "purchase_name": "大化县财政局",
"source_code": "test", "source_name": "test"},
{"title": "南宁市采购公告", "purchase_name": "南宁市财政局",
"source_code": "test", "source_name": "test"},
]
result = filter_by_keywords(announcements, ["大化"])
assert len(result) == 1
assert result[0]["title"] == "大化县采购公告"
def test_filter_by_keywords_no_keywords():
announcements = [
{"title": "大化县采购公告", "purchase_name": "x",
"source_code": "test", "source_name": "test"},
]
result = filter_by_keywords(announcements, [])
assert len(result) == 1
def test_filter_by_date_range():
today = datetime(2026, 5, 9)
announcements = [
{"title": "t1", "publish_date": datetime(2026, 5, 9),
"source_code": "test", "source_name": "test"},
{"title": "t2", "publish_date": datetime(2026, 5, 1),
"source_code": "test", "source_name": "test"},
{"title": "t3", "publish_date": datetime(2026, 4, 30),
"source_code": "test", "source_name": "test"},
]
result = filter_by_date(announcements, start_date=today, end_date=today)
assert len(result) == 1
assert result[0]["title"] == "t1"
```
- [ ] **Step 2: 创建管道测试 tests/test_services/test_pipeline.py**
```python
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from datetime import datetime
from app.services.pipeline import PostCrawlPipeline
from app.crawler.base import CrawlResult, PipelineConfig
@pytest.mark.asyncio
async def test_pipeline_filtered_mode():
config = PipelineConfig(
filter_enabled=True,
keywords=["大化"],
dedup_enabled=True,
notify_mode="filtered",
mark_sent=False,
)
result = CrawlResult(
source_code="test", source_name="test",
total_count=3, new_count=3,
announcements=[
{
"title": "大化县公告", "publish_date": datetime(2026, 5, 9),
"purchase_name": "大化县", "content_url": "https://1.com",
"source_code": "test", "source_name": "测试",
"announcement_type": "purchase", "crawl_mode": "auto",
"is_new": True, "is_today": True,
"content_hash": "abc123",
},
{
"title": "南宁市公告", "publish_date": datetime(2026, 5, 9),
"purchase_name": "南宁市", "content_url": "https://2.com",
"source_code": "test", "source_name": "测试",
"announcement_type": "purchase", "crawl_mode": "auto",
"is_new": True, "is_today": True,
"content_hash": "def456",
},
],
)
mock_db = AsyncMock()
mock_notify = AsyncMock()
pipeline = PostCrawlPipeline(db_session=mock_db, notification_service=mock_notify)
with patch.object(pipeline, "_save_to_db", AsyncMock(return_value=2)):
with patch.object(pipeline, "_send_notifications", AsyncMock(return_value=1)):
pipe_result = await pipeline.process(
result.announcements, config
)
assert pipe_result.stored == 2
assert pipe_result.filtered == 1 # 3 total, 2 pass keyword, 1 filtered
assert pipe_result.notified == 1
@pytest.mark.asyncio
async def test_pipeline_all_mode():
config = PipelineConfig(
filter_enabled=False,
keywords=[],
dedup_enabled=True,
notify_mode="all",
mark_sent=True,
)
result = CrawlResult(
source_code="dahuagov", source_name="大化县政府网",
total_count=2, new_count=2,
announcements=[
{
"title": f"公告{i}", "publish_date": datetime(2026, 5, 9),
"purchase_name": "大化县", "content_url": f"https://x.com/{i}",
"source_code": "dahuagov", "source_name": "大化县政府网采购公告",
"announcement_type": "purchase", "crawl_mode": "auto",
"is_new": True, "is_today": True,
"content_hash": f"hash{i}",
}
for i in range(2)
],
)
mock_db = AsyncMock()
mock_notify = AsyncMock()
pipeline = PostCrawlPipeline(db_session=mock_db, notification_service=mock_notify)
with patch.object(pipeline, "_save_to_db", AsyncMock(return_value=2)):
with patch.object(pipeline, "_send_notifications", AsyncMock(return_value=2)):
with patch.object(pipeline, "_mark_sent", AsyncMock(return_value=2)):
pipe_result = await pipeline.process(
result.announcements, config
)
assert pipe_result.stored == 2
assert pipe_result.filtered == 0
assert pipe_result.notified == 2
pipeline._mark_sent.assert_awaited_once()
```
- [ ] **Step 3: 运行测试确认失败**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_services/test_filter_service.py tests/test_services/test_pipeline.py -v
# Expected: FAIL — ModuleNotFoundError
```
- [ ] **Step 4: 创建 app/services/filter_service.py**
```python
from datetime import date
from typing import Any, Dict, List, Optional
def filter_by_keywords(announcements: List[Dict[str, Any]],
keywords: List[str]) -> List[Dict[str, Any]]:
if not keywords:
return announcements
filtered = []
for ann in announcements:
search_text = f"{ann.get('title', '')} {ann.get('purchase_name', '')}"
if any(kw in search_text for kw in keywords):
ann["keyword_matched"] = True
filtered.append(ann)
else:
ann["keyword_matched"] = False
filtered.append(ann) # Still include, just mark not matched
return filtered
def filter_by_date(announcements: List[Dict[str, Any]],
start_date: Optional[date] = None,
end_date: Optional[date] = None) -> List[Dict[str, Any]]:
if not start_date and not end_date:
return announcements
filtered = []
for ann in announcements:
pub_date = ann.get("publish_date")
if not pub_date:
continue
if isinstance(pub_date, date):
pub_date = pub_date
else:
pub_date = pub_date.date() if hasattr(pub_date, "date") else pub_date
if start_date and pub_date < start_date:
continue
if end_date and pub_date > end_date:
continue
filtered.append(ann)
return filtered
def dedup_by_hash(announcements: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
seen = set()
result = []
for ann in announcements:
h = ann.get("content_hash")
if h and h not in seen:
seen.add(h)
result.append(ann)
return result
```
- [ ] **Step 5: 创建 app/services/pipeline.py**
```python
from typing import Any, Dict, List
from app.crawler.base import PipelineConfig, PipelineResult
from app.services.filter_service import filter_by_keywords, dedup_by_hash
class PostCrawlPipeline:
def __init__(self, db_session, notification_service):
self.db = db_session
self.notify = notification_service
async def process(self, announcements: List[Dict[str, Any]],
config: PipelineConfig) -> PipelineResult:
result = PipelineResult()
if not announcements:
return result
# 1. 去重
if config.dedup_enabled:
announcements = dedup_by_hash(announcements)
# 2. 存储到数据库
stored = await self._save_to_db(announcements)
result.stored = stored
to_notify = announcements
# 3. 筛选
if config.filter_enabled and config.keywords:
before = len(to_notify)
to_notify = [a for a in to_notify
if self._match_keywords(a, config.keywords)]
result.filtered = before - len(to_notify)
# 4. 推送
if config.notify_mode == "all":
result.notified = await self._send_notifications(to_notify)
elif config.notify_mode == "filtered":
if config.filter_enabled and config.keywords:
result.notified = await self._send_notifications(to_notify)
elif not config.filter_enabled:
result.notified = await self._send_notifications(to_notify)
# 5. 标记已发送
if config.mark_sent and result.notified > 0:
await self._mark_sent(to_notify)
return result
async def _save_to_db(self, announcements: List[Dict[str, Any]]) -> int:
from sqlalchemy.dialects.postgresql import insert
from app.models.announcement import Announcement
if not announcements:
return 0
values = [{
"title": a["title"],
"publish_date": a["publish_date"],
"purchase_name": a.get("purchase_name", ""),
"content_url": a.get("content_url", ""),
"source_code": a["source_code"],
"source_name": a["source_name"],
"announcement_type": a.get("announcement_type", "purchase"),
"content_hash": a["content_hash"],
"crawl_mode": a.get("crawl_mode", "auto"),
"is_new": a.get("is_new", True),
"is_sent": False,
"keyword_matched": a.get("keyword_matched", False),
} for a in announcements]
stmt = insert(Announcement).values(values)
stmt = stmt.on_conflict_do_nothing(index_elements=["content_hash"])
result_proxy = await self.db.execute(stmt)
await self.db.commit()
return result_proxy.rowcount or len(values)
async def _send_notifications(self, announcements: List[Dict[str, Any]]) -> int:
return await self.notify.send(announcements)
async def _mark_sent(self, announcements: List[Dict[str, Any]]) -> int:
from app.models.announcement import Announcement
from sqlalchemy import update
hashes = [a["content_hash"] for a in announcements if a.get("content_hash")]
if not hashes:
return 0
stmt = (
update(Announcement)
.where(Announcement.content_hash.in_(hashes))
.values(is_sent=True)
)
result = await self.db.execute(stmt)
await self.db.commit()
return result.rowcount
@staticmethod
def _match_keywords(announcement: Dict[str, Any], keywords: List[str]) -> bool:
text = f"{announcement.get('title', '')} {announcement.get('purchase_name', '')}"
return any(kw in text for kw in keywords)
```
- [ ] **Step 6: 运行测试验证通过**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_services/ -v
# Expected: all passed
```
- [ ] **Step 7: 运行全部测试**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/ -v
# Expected: all passed
```
- [ ] **Step 8: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/services/pipeline.py app/services/filter_service.py tests/test_services/
git commit -m "feat: 添加 PostCrawlPipeline 统一管道 + 筛选服务 + 测试"
```
---
### Task 9: 企业微信模块
**Files:**
- Create: `app/wechat/__init__.py`
- Create: `app/wechat/crypto.py` — 复制现有 WXBizMsgCrypt
- Create: `app/wechat/client.py` — 企业微信 API 客户端
- Create: `app/wechat/handler.py` — 消息处理器
- [ ] **Step 1: 创建 app/wechat/__init__.py** (空文件)
- [ ] **Step 2: 创建 app/wechat/crypto.py**
从现有 `gx_gp_monitor/wechat/WXBizMsgCrypt.py` 和 `gx_gp_monitor/wechat/ierror.py` 复制并整合：
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
cp gx_gp_monitor/wechat/WXBizMsgCrypt.py app/wechat/crypto.py
cp gx_gp_monitor/wechat/ierror.py app/wechat/ierror.py
```
- [ ] **Step 3: 创建 app/wechat/client.py**
```python
import time
import httpx
from typing import Optional
from app.config import settings
class WeChatClient:
def __init__(self):
self._access_token: Optional[str] = None
self._token_expires_at: float = 0
async def _get_access_token(self) -> Optional[str]:
now = time.time()
if self._access_token and now < self._token_expires_at:
return self._access_token
url = "https://qyapi.weixin.qq.com/cgi-bin/gettoken"
params = {
"corpid": settings.wechat_corp_id,
"corpsecret": settings.wechat_secret,
}
async with httpx.AsyncClient(timeout=30) as client:
response = await client.get(url, params=params)
data = response.json()
if data.get("errcode") == 0:
self._access_token = data["access_token"]
self._token_expires_at = now + data.get("expires_in", 7200) - 300
return self._access_token
return None
async def send_text(self, content: str, to_user: str = "@all") -> bool:
return await self._send_message("text", {"content": content}, to_user)
async def send_markdown(self, content: str, to_user: str = "@all") -> bool:
return await self._send_message("markdown", {"content": content}, to_user)
async def send_textcard(self, title: str, description: str, url: str,
to_user: str = "@all", btn_txt: str = "查看详情") -> bool:
return await self._send_message("textcard", {
"title": title,
"description": description,
"url": url,
"btntxt": btn_txt,
}, to_user)
async def _send_message(self, msgtype: str, msg_data: dict,
to_user: str = "@all") -> bool:
token = await self._get_access_token()
if not token:
return False
url = "https://qyapi.weixin.qq.com/cgi-bin/message/send"
params = {"access_token": token}
body = {
"touser": to_user,
"msgtype": msgtype,
"agentid": int(settings.wechat_agent_id),
msgtype: msg_data,
}
async with httpx.AsyncClient(timeout=30) as client:
response = await client.post(url, params=params, json=body)
data = response.json()
return data.get("errcode") == 0
```
- [ ] **Step 4: 创建 app/wechat/handler.py**
```python
import xml.etree.cElementTree as ET
from typing import Optional
from app.wechat.crypto import WXBizMsgCrypt
from app.config import settings
class WeChatMessageHandler:
def __init__(self):
self.wxcpt = WXBizMsgCrypt(
sToken=settings.wechat_token,
sEncodingAESKey=settings.wechat_encoding_aes_key,
sReceiveId=settings.wechat_corp_id,
)
def verify_url(self, msg_signature: str, timestamp: str,
nonce: str, echostr: str) -> Optional[str]:
ret, sEchoStr = self.wxcpt.VerifyURL(msg_signature, timestamp, nonce, echostr)
if ret == 0:
return sEchoStr.decode("utf-8") if isinstance(sEchoStr, bytes) else sEchoStr
return None
def decrypt_message(self, post_data: str, msg_signature: str,
timestamp: str, nonce: str) -> Optional[ET.Element]:
ret, xml_content = self.wxcpt.DecryptMsg(
post_data, msg_signature, timestamp, nonce
)
if ret != 0:
return None
return ET.fromstring(xml_content)
def encrypt_response(self, response_xml: str, nonce: str,
timestamp: str) -> Optional[str]:
ret, encrypted = self.wxcpt.EncryptMsg(response_xml, nonce, timestamp)
if ret == 0:
return encrypted
return None
def handle_event(self, event: str, event_key: Optional[str],
from_user: str) -> Optional[str]:
return None
def handle_text(self, content: str, from_user: str) -> Optional[str]:
return None
```
- [ ] **Step 5: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/wechat/
git commit -m "feat: 迁移企业微信模块（crypto + client + handler）"
```
---
### Task 10: Notification Service
**Files:**
- Create: `app/services/notification_service.py`
- Test: `tests/test_services/test_notification_service.py`
- [ ] **Step 1: 创建测试 tests/test_services/test_notification_service.py**
```python
import pytest
from unittest.mock import AsyncMock, patch
from app.services.notification_service import NotificationService
@pytest.mark.asyncio
async def test_notification_send():
svc = NotificationService()
announcements = [
{
"title": "测试公告",
"publish_date": None,
"purchase_name": "测试单位",
"content_url": "https://x.com/1",
"source_code": "test",
"source_name": "测试来源",
"announcement_type": "purchase",
}
]
with patch.object(svc.client, "send_textcard", AsyncMock(return_value=True)):
count = await svc.send(announcements)
assert count == 1
@pytest.mark.asyncio
async def test_notification_send_empty():
svc = NotificationService()
count = await svc.send([])
assert count == 0
@pytest.mark.asyncio
async def test_notification_send_disabled():
svc = NotificationService()
announcements = [{"title": "test"}]
with patch.object(svc.client, "send_textcard", AsyncMock(return_value=False)):
count = await svc.send(announcements)
assert count == 0
```
- [ ] **Step 2: 运行测试确认失败**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_services/test_notification_service.py -v
# Expected: FAIL
```
- [ ] **Step 3: 创建 app/services/notification_service.py**
```python
from datetime import datetime
from typing import Any, Dict, List
from app.config import settings
from app.wechat.client import WeChatClient
class NotificationService:
def __init__(self):
self.client = WeChatClient()
async def send(self, announcements: List[Dict[str, Any]]) -> int:
if not settings.wechat_enabled:
return 0
if not announcements:
return 0
sent = 0
for ann in announcements:
try:
title = ann.get("title", "")
if len(title) > 128:
title = title[:125] + "..."
purchase_name = ann.get("purchase_name", "")
if len(purchase_name) > 25:
purchase_name = purchase_name[:22] + "..."
pub_date = ann.get("publish_date")
time_str = pub_date.strftime("%Y-%m-%d %H:%M") if pub_date else "时间未知"
source_name = ann.get("source_name", "")
description = (
f'<div style="font-size: 14px; margin-top: 8px;">'
f'{source_name} | {purchase_name} | {time_str}'
f'</div>'
)
url = ann.get("content_url", "")
if await self.client.send_textcard(title, description, url):
sent += 1
except Exception:
continue
return sent
```
- [ ] **Step 4: 运行测试验证通过**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_services/test_notification_service.py -v
# Expected: 3 passed
```
- [ ] **Step 5: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/services/notification_service.py tests/test_services/test_notification_service.py
git commit -m "feat: 添加 NotificationService 通知服务 + 测试"
```
---
### Task 11: API 路由
**Files:**
- Create: `app/api/__init__.py`
- Create: `app/api/deps.py`
- Create: `app/api/router.py`
- Create: `app/api/announcements.py`
- Create: `app/api/crawl.py`
- Create: `app/api/wechat.py`
- Modify: `app/main.py` — 注册路由
- Test: `tests/test_api/__init__.py`
- Test: `tests/test_api/test_health.py`
- Test: `tests/test_api/test_announcements.py`
- Test: `tests/test_api/test_crawl.py`
- [ ] **Step 1: 创建 app/api/__init__.py** (空文件)
- [ ] **Step 2: 创建 app/api/deps.py**
```python
from sqlalchemy.ext.asyncio import AsyncSession
from app.main import async_session
from app.services.crawl_service import CrawlService
from app.crawler.gxgp_spider import GXGPSpider
from app.crawler.dahuagov_spider import DahuagovSpider
async def get_db() -> AsyncSession:
async with async_session() as session:
yield session
_crawl_service: CrawlService | None = None
def get_crawl_service() -> CrawlService:
global _crawl_service
if _crawl_service is None:
_crawl_service = CrawlService()
_crawl_service.register(GXGPSpider())
_crawl_service.register(DahuagovSpider())
return _crawl_service
```
- [ ] **Step 3: 创建 app/api/router.py**
```python
from fastapi import APIRouter
from app.api import announcements, crawl, wechat
api_router = APIRouter(prefix="/api/v1")
api_router.include_router(announcements.router, tags=["announcements"])
api_router.include_router(crawl.router, tags=["crawl"])
api_router.include_router(wechat.router, tags=["wechat"])
```
- [ ] **Step 4: 创建 app/api/announcements.py**
```python
from datetime import datetime
from typing import Optional
from fastapi import APIRouter, Depends, Query, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, text
from app.api.deps import get_db
from app.models.announcement import Announcement
from app.models.schemas import AnnouncementResponse, AnnouncementListResponse
router = APIRouter()
@router.get("/announcements", response_model=AnnouncementListResponse)
async def list_announcements(
page: int = Query(1, ge=1),
page_size: int = Query(20, ge=1, le=100),
source_code: Optional[str] = None,
keyword: Optional[str] = None,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
crawl_mode: Optional[str] = None,
db: AsyncSession = Depends(get_db),
):
conditions = []
if source_code:
conditions.append(Announcement.source_code == source_code)
if crawl_mode:
conditions.append(Announcement.crawl_mode == crawl_mode)
if start_date:
conditions.append(Announcement.publish_date >= start_date)
if end_date:
conditions.append(Announcement.publish_date <= end_date)
if keyword:
conditions.append(
Announcement.title.ilike(f"%{keyword}%")
)
base_query = select(Announcement)
if conditions:
base_query = base_query.where(*conditions)
count_query = select(func.count()).select_from(base_query.subquery())
total_result = await db.execute(count_query)
total = total_result.scalar() or 0
items_query = base_query.order_by(Announcement.publish_date.desc()) \
.offset((page - 1) * page_size).limit(page_size)
items_result = await db.execute(items_query)
items = items_result.scalars().all()
return AnnouncementListResponse(
total=total,
page=page,
page_size=page_size,
items=[AnnouncementResponse.model_validate(item) for item in items],
)
@router.get("/announcements/{announcement_id}", response_model=AnnouncementResponse)
async def get_announcement(announcement_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(
select(Announcement).where(Announcement.id == announcement_id)
)
item = result.scalar_one_or_none()
if item is None:
raise HTTPException(status_code=404, detail="公告不存在")
return AnnouncementResponse.model_validate(item)
@router.get("/announcements/today", response_model=AnnouncementListResponse)
async def get_today_announcements(db: AsyncSession = Depends(get_db)):
today = datetime.now().date()
result = await db.execute(
select(Announcement).where(
func.date(Announcement.publish_date) == today
).order_by(Announcement.publish_date.desc())
)
items = result.scalars().all()
return AnnouncementListResponse(
total=len(items), page=1, page_size=len(items),
items=[AnnouncementResponse.model_validate(item) for item in items],
)
@router.get("/announcements/stats")
async def get_stats(db: AsyncSession = Depends(get_db)):
total = await db.execute(select(func.count()).select_from(Announcement))
today_count = await db.execute(
select(func.count()).where(
func.date(Announcement.publish_date) == func.current_date()
).select_from(Announcement)
)
new_count = await db.execute(
select(func.count()).where(Announcement.is_new == True)
.select_from(Announcement)
)
unsent = await db.execute(
select(func.count()).where(
Announcement.is_sent == False, Announcement.is_new == True
).select_from(Announcement)
)
return {
"total": total.scalar() or 0,
"today": today_count.scalar() or 0,
"new": new_count.scalar() or 0,
"unsent": unsent.scalar() or 0,
}
```
- [ ] **Step 5: 创建 app/api/crawl.py**
```python
from fastapi import APIRouter, Depends
from app.api.deps import get_crawl_service
from app.models.schemas import CrawlTriggerRequest
router = APIRouter()
@router.post("/crawl/trigger")
async def trigger_crawl(request: CrawlTriggerRequest):
from app.services.pipeline import PostCrawlPipeline
from app.services.notification_service import NotificationService
from app.api.deps import get_db
import asyncio
service = get_crawl_service()
names = service.get_spider_names()
all_results = []
for name in names:
results = await service.run_spider(name)
all_results.extend(results)
return {
"spiders_run": names,
"total_announcements": sum(r.total_count for r in all_results),
"errors": [r.error_message for r in all_results if not r.success],
}
@router.get("/crawl/status")
async def crawl_status():
service = get_crawl_service()
return {
"spiders": service.get_spider_names(),
"running": False,
}
@router.get("/crawl/sources")
async def crawl_sources():
import json
from app.config import settings
sources = json.loads(settings.announcement_sources)
return {
"sources": [
{"code": code, "name": info["name"], "type": info["type"]}
for code, info in sources.items()
]
}
```
- [ ] **Step 6: 创建 app/api/wechat.py**
```python
from fastapi import APIRouter, Request, Response
from app.wechat.handler import WeChatMessageHandler
router = APIRouter()
_handler: WeChatMessageHandler | None = None
def get_handler() -> WeChatMessageHandler:
global _handler
if _handler is None:
_handler = WeChatMessageHandler()
return _handler
@router.get("/wechat/callback")
async def wechat_verify(request: Request):
handler = get_handler()
params = request.query_params
echostr = handler.verify_url(
params.get("msg_signature", ""),
params.get("timestamp", ""),
params.get("nonce", ""),
params.get("echostr", ""),
)
if echostr:
return Response(content=echostr, media_type="text/plain")
return Response(content="verification failed", status_code=403)
@router.post("/wechat/callback")
async def wechat_callback(request: Request):
handler = get_handler()
params = request.query_params
post_data = await request.body()
post_text = post_data.decode("utf-8")
xml_tree = handler.decrypt_message(
post_text,
params.get("msg_signature", ""),
params.get("timestamp", ""),
params.get("nonce", ""),
)
if xml_tree is None:
return Response(content="decrypt failed", status_code=403)
msg_type = xml_tree.find("MsgType")
msg_type = msg_type.text if msg_type is not None else "unknown"
if msg_type == "event":
event = xml_tree.find("Event")
event_key = xml_tree.find("EventKey")
from_user = xml_tree.find("FromUserName")
handler.handle_event(
event.text if event is not None else "",
event_key.text if event_key is not None else None,
from_user.text if from_user is not None else "",
)
elif msg_type == "text":
content = xml_tree.find("Content")
from_user = xml_tree.find("FromUserName")
handler.handle_text(
content.text if content is not None else "",
from_user.text if from_user is not None else "",
)
return Response(content="success")
```
- [ ] **Step 7: 创建 app/api/scheduler.py** (scheduler endpoints 在 spec 4.3)
```python
from fastapi import APIRouter
from app.scheduler.jobs import scheduler
from app.models.schemas import JobResponse
router = APIRouter(prefix="/scheduler", tags=["scheduler"])
@router.get("/jobs")
async def list_jobs():
jobs = []
for job in scheduler.get_jobs():
jobs.append(JobResponse(
id=job.id,
name=job.name,
next_run_time=str(job.next_run_time) if job.next_run_time else None,
))
return {"jobs": jobs}
@router.post("/pause/{job_id}")
async def pause_job(job_id: str):
scheduler.pause_job(job_id)
return {"status": "paused", "job_id": job_id}
@router.post("/resume/{job_id}")
async def resume_job(job_id: str):
scheduler.resume_job(job_id)
return {"status": "resumed", "job_id": job_id}
```
- [ ] **Step 8: 修改 app/api/router.py 注册 scheduler 路由**
在 `app/api/router.py` 中添加:
```python
from app.api import scheduler as scheduler_module
api_router.include_router(scheduler_module.router)
```
- [ ] **Step 9: 修改 app/main.py 注册路由**
在 app/main.py 的 `app = FastAPI(...)` 之后添加：
```python
from app.api.router import api_router
app.include_router(api_router)
```
- [ ] **Step 10: 创建测试 tests/test_api/__init__.py** (空文件)
- [ ] **Step 11: 创建测试 tests/test_api/test_health.py**
```python
import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app
@pytest.mark.asyncio
async def test_health():
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
response = await client.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
```
- [ ] **Step 12: 运行 API 测试**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/test_api/test_health.py -v
# Expected: 1 passed
```
- [ ] **Step 13: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/api/ app/main.py tests/test_api/
git commit -m "feat: 添加 API 路由（公告/爬取/微信回调/健康检查）+ 测试"
```
---
### Task 12: APScheduler 定时任务
**Files:**
- Create: `app/scheduler/__init__.py`
- Create: `app/scheduler/jobs.py`
- Modify: `app/main.py` — lifespan 中启动调度器
- [ ] **Step 1: 创建 app/scheduler/__init__.py** (空文件)
- [ ] **Step 2: 创建 app/scheduler/jobs.py**
```python
import logging
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from app.config import settings
from app.api.deps import get_crawl_service
logger = logging.getLogger(__name__)
scheduler = AsyncIOScheduler()
async def scheduled_crawl():
"""定时爬取任务：运行所有 Spider"""
logger.info("开始定时爬取任务")
service = get_crawl_service()
names = service.get_spider_names()
for name in names:
try:
results = await service.run_spider(name)
for r in results:
if not r.success:
logger.error(f"Spider {name} 失败: {r.error_message}")
else:
logger.info(f"Spider {name} 完成: {r.total_count} 条")
except Exception as e:
logger.error(f"Spider {name} 异常: {e}")
logger.info("定时爬取任务完成")
def start_scheduler():
if not settings.scheduler_enabled:
return
scheduler.add_job(
scheduled_crawl,
"cron",
hour="8,14,18",
minute="0",
id="scheduled_crawl",
name="定时爬取",
timezone="Asia/Shanghai",
)
scheduler.start()
logger.info("APScheduler 已启动 (8:00, 14:00, 18:00)")
def shutdown_scheduler():
if scheduler.running:
scheduler.shutdown(wait=False)
logger.info("APScheduler 已停止")
```
- [ ] **Step 3: 修改 app/main.py 的 lifespan**
```python
@asynccontextmanager
async def lifespan(app: FastAPI):
import logging
logging.basicConfig(
level=getattr(logging, settings.log_level),
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
from app.scheduler.jobs import start_scheduler, shutdown_scheduler
start_scheduler()
yield
shutdown_scheduler()
await engine.dispose()
```
- [ ] **Step 4: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add app/scheduler/ app/main.py
git commit -m "feat: 添加 APScheduler 定时爬取任务"
```
---
### Task 13: Docker 化
**Files:**
- Create: `docker/Dockerfile`
- Create: `docker/docker-compose.yml`
- [ ] **Step 1: 创建 docker/Dockerfile**
```dockerfile
FROM python:3.12-slim
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \
libpq-dev && \
rm -rf /var/lib/apt/lists/*
COPY pyproject.toml .
RUN pip install --no-cache-dir -e ".[dev]"
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
```
- [ ] **Step 2: 创建 docker/docker-compose.yml**
```yaml
services:
app:
build:
context: ..
dockerfile: docker/Dockerfile
ports:
- "8000:8000"
env_file:
- ../.env
volumes:
- ../logs:/app/logs
restart: unless-stopped
```
- [ ] **Step 3: 构建验证**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
docker build -f docker/Dockerfile -t gx-gp-notify .
# Expected: Successfully built
```
- [ ] **Step 4: 提交**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add docker/
git commit -m "feat: 添加 Docker 部署配置"
```
---
### Task 14: 运行 ruff 代码检查
- [ ] **Step 1: 安装 ruff 并运行**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
pip install ruff
ruff check app/ tests/
# Fix any issues
ruff check --fix app/ tests/
```
- [ ] **Step 2: 提交修复**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add -A
git commit -m "chore: ruff 代码检查与修复"
```
---
### Task 15: 清理旧文件并最终验证
- [ ] **Step 1: 运行全部测试最终确认**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
python -m pytest tests/ -v
# Expected: all passed
```
- [ ] **Step 2: 验证服务启动**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
uvicorn app.main:app --host 0.0.0.0 --port 8000 &
sleep 2
curl http://localhost:8000/health
curl http://localhost:8000/docs # 需要 debug=true
kill %1
```
- [ ] **Step 3: 提交最终状态**
```bash
cd /home/v6ole/PythonProject/GX-gp-notify
git add -A
git commit -m "chore: 清理旧文件，最终验证通过"
```
Reference in New Issue View Git Blame Copy Permalink