我来为你设计一个完整的PHP数据采集与发布系统,这个系统将包含采集、处理、发布的核心功能。

系统架构设计
项目结构
data-collection-system/
├── config/
│ └── config.php
├── core/
│ ├── Database.php
│ ├── Collector.php
│ ├── Parser.php
│ ├── Publisher.php
│ └── Validator.php
├── collectors/
│ ├── HtmlCollector.php
│ ├── ApiCollector.php
│ └── RssCollector.php
├── publishers/
│ ├── DatabasePublisher.php
│ ├── FilePublisher.php
│ └── ApiPublisher.php
├── cron/
│ └── run_collector.php
└── logs/
└── system.log
核心配置类
<?php
// config/config.php
return [
'database' => [
'host' => 'localhost',
'dbname' => 'collection_system',
'username' => 'root',
'password' => '',
'charset' => 'utf8mb4'
],
'curl' => [
'timeout' => 30,
'user_agent' => 'DataCollector/1.0',
'proxy' => null,
'ssl_verify' => false
],
'log' => [
'path' => __DIR__ . '/../logs/',
'level' => 'DEBUG'
],
'publish' => [
'type' => 'database', // database, file, api
'batch_size' => 100
]
];
数据库连接类
<?php
// core/Database.php
class Database {
private static $instance = null;
private $connection;
private function __construct() {
$config = require __DIR__ . '/../config/config.php';
try {
$dsn = "mysql:host={$config['database']['host']};dbname={$config['database']['dbname']};charset={$config['database']['charset']}";
$this->connection = new PDO($dsn, $config['database']['username'], $config['database']['password']);
$this->connection->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION);
$this->connection->setAttribute(PDO::ATTR_DEFAULT_FETCH_MODE, PDO::FETCH_ASSOC);
} catch (PDOException $e) {
throw new Exception("Database connection failed: " . $e->getMessage());
}
}
public static function getInstance() {
if (self::$instance === null) {
self::$instance = new self();
}
return self::$instance;
}
public function getConnection() {
return $this->connection;
}
public function query($sql, $params = []) {
$stmt = $this->connection->prepare($sql);
$stmt->execute($params);
return $stmt;
}
public function insert($table, $data) {
$columns = implode(', ', array_keys($data));
$placeholders = ':' . implode(', :', array_keys($data));
$sql = "INSERT INTO $table ($columns) VALUES ($placeholders)";
return $this->query($sql, $data);
}
}
采集器基类
<?php
// core/Collector.php
abstract class Collector {
protected $config;
protected $logger;
protected $parser;
public function __construct() {
$this->config = require __DIR__ . '/../config/config.php';
$this->logger = new Logger();
$this->parser = new Parser();
}
abstract public function collect($params);
protected function validateData($data) {
return Validator::validate($data);
}
protected function log($message, $level = 'INFO') {
$this->logger->log($message, $level);
}
}
HTML采集器
<?php
// collectors/HtmlCollector.php
class HtmlCollector extends Collector {
public function collect($params) {
$url = $params['url'];
$selector = $params['selector'] ?? 'body';
$html = $this->fetchHtml($url);
if (empty($html)) {
$this->log("Failed to fetch HTML from: $url", 'ERROR');
return [];
}
$data = $this->parser->parseHtml($html, $selector);
$validatedData = $this->validateData($data);
return $validatedData;
}
private function fetchHtml($url) {
$ch = curl_init();
curl_setopt_array($ch, [
CURLOPT_URL => $url,
CURLOPT_RETURNTRANSFER => true,
CURLOPT_FOLLOWLOCATION => true,
CURLOPT_TIMEOUT => $this->config['curl']['timeout'],
CURLOPT_USERAGENT => $this->config['curl']['user_agent'],
CURLOPT_SSL_VERIFYPEER => $this->config['curl']['ssl_verify']
]);
$response = curl_exec($ch);
$error = curl_error($ch);
curl_close($ch);
if ($error) {
$this->log("CURL Error: $error", 'ERROR');
return '';
}
return $response;
}
}
解析器类
<?php
// core/Parser.php
class Parser {
public function parseHtml($html, $selector) {
$dom = new DOMDocument();
@$dom->loadHTML($html);
$xpath = new DOMXPath($dom);
$elements = $xpath->query($selector);
$data = [];
foreach ($elements as $element) {
$item = [
'title' => $this->extractTitle($element),
'content' => $this->extractContent($element),
'image' => $this->extractImage($element),
'url' => $this->extractUrl($element),
'timestamp' => time()
];
$data[] = $item;
}
return $data;
}
private function extractTitle($element) {
$titleNode = $element->getElementsByTagName('title')->item(0);
return $titleNode ? $titleNode->textContent : '';
}
private function extractContent($element) {
$contentNode = $element->getElementsByTagName('content')->item(0);
if ($contentNode) {
return $contentNode->textContent;
}
// 提取文章正文
$paragraphs = $element->getElementsByTagName('p');
$content = '';
foreach ($paragraphs as $p) {
$content .= $p->textContent . "\n";
}
return trim($content);
}
private function extractImage($element) {
$imgNode = $element->getElementsByTagName('img')->item(0);
return $imgNode ? $imgNode->getAttribute('src') : '';
}
private function extractUrl($element) {
$linkNode = $element->getElementsByTagName('a')->item(0);
return $linkNode ? $linkNode->getAttribute('href') : '';
}
public function parseJson($json) {
return json_decode($json, true);
}
}
发布器基类
<?php
// core/Publisher.php
abstract class Publisher {
protected $config;
protected $logger;
public function __construct() {
$this->config = require __DIR__ . '/../config/config.php';
$this->logger = new Logger();
}
abstract public function publish($data);
protected function validateForPublish($data) {
return true; // 发布前的验证逻辑
}
}
数据库发布器
<?php
// publishers/DatabasePublisher.php
class DatabasePublisher extends Publisher {
private $db;
public function __construct() {
parent::__construct();
$this->db = Database::getInstance()->getConnection();
}
public function publish($data) {
if (empty($data)) {
return false;
}
$publishedCount = 0;
foreach ($data as $item) {
if ($this->validateForPublish($item)) {
$this->checkAndCreateTables();
$result = $this->insertContent($item);
if ($result) {
$publishedCount++;
}
}
}
$this->logger->log("Published $publishedCount items to database");
return $publishedCount;
}
private function checkAndCreateTables() {
$sql = "CREATE TABLE IF NOT EXISTS collected_data (
id INT PRIMARY KEY AUTO_INCREMENT,
title VARCHAR(500),
content TEXT,
image_url VARCHAR(500),
source_url VARCHAR(500),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY unique_url (source_url)
)";
$this->db->exec($sql);
}
private function insertContent($data) {
$sql = "INSERT INTO collected_data (title, content, image_url, source_url)
VALUES (:title, :content, :image_url, :source_url)
ON DUPLICATE KEY UPDATE id=id";
$stmt = $this->db->prepare($sql);
return $stmt->execute([
':title' => $data['title'],
':content' => $data['content'],
':image_url' => $data['image'] ?? '',
':source_url' => $data['url'] ?? ''
]);
}
}
验证器类
<?php
// core/Validator.php
class Validator {
public static function validate($data) {
$validated = [];
foreach ($data as $item) {
if (self::isValidItem($item)) {
$validated[] = $item;
}
}
return $validated;
}
private static function isValidItem($item) {
// 检查必要字段
if (empty($item['title']) || empty($item['content'])) {
return false;
}
// 检查内容长度
if (strlen($item['content']) < 50) {
return false;
}
// 检查重复内容
if (self::isDuplicate($item)) {
return false;
}
return true;
}
private static function isDuplicate($item) {
$db = Database::getInstance()->getConnection();
$stmt = $db->prepare("SELECT COUNT(*) FROM collected_data WHERE source_url = :url");
$stmt->execute([':url' => $item['url'] ?? '']);
return $stmt->fetchColumn() > 0;
}
public static function clean($data) {
// 清理非法字符
$data = strip_tags($data);
$data = htmlspecialchars($data, ENT_QUOTES, 'UTF-8');
$data = trim($data);
return $data;
}
}
日志类
<?php
// core/Logger.php
class Logger {
private $logPath;
private $level;
public function __construct() {
$config = require __DIR__ . '/../config/config.php';
$this->logPath = $config['log']['path'];
$this->level = $config['log']['level'];
if (!file_exists($this->logPath)) {
mkdir($this->logPath, 0755, true);
}
}
public function log($message, $level = 'INFO') {
$trace = debug_backtrace();
$caller = $trace[1]['class'] ?? 'System';
$method = $trace[1]['function'] ?? 'unknown';
$logEntry = sprintf(
"[%s] [%s] [%s::%s] %s%s",
date('Y-m-d H:i:s'),
$level,
$caller,
$method,
$message,
PHP_EOL
);
$filename = $this->logPath . date('Y-m-d') . '.log';
file_put_contents($filename, $logEntry, FILE_APPEND | LOCK_EX);
}
}
定时任务脚本
<?php
// cron/run_collector.php
require_once __DIR__ . '/../core/Database.php';
require_once __DIR__ . '/../core/Collector.php';
require_once __DIR__ . '/../core/Parser.php';
require_once __DIR__ . '/../core/Publisher.php';
require_once __DIR__ . '/../core/Validator.php';
require_once __DIR__ . '/../core/Logger.php';
class RunCollector {
private $logger;
public function __construct() {
$this->logger = new Logger();
}
public function execute($collectorType = 'html') {
$this->logger->log("Starting collection process", 'INFO');
try {
// 选择采集器
$collector = $this->getCollector($collectorType);
// 配置采集参数
$params = $this->getCollectionParams($collectorType);
// 执行采集
$data = $collector->collect($params);
$this->logger->log("Collected " . count($data) . " items", 'INFO');
// 选择发布器
$publisher = $this->getPublisher();
// 执行发布
$published = $publisher->publish($data);
$this->logger->log("Published $published items", 'INFO');
return $published;
} catch (Exception $e) {
$this->logger->log("Collection failed: " . $e->getMessage(), 'ERROR');
return 0;
}
}
private function getCollector($type) {
switch ($type) {
case 'html':
return new HtmlCollector();
case 'api':
return new ApiCollector();
case 'rss':
return new RssCollector();
default:
throw new Exception("Unknown collector type: $type");
}
}
private function getPublisher() {
$config = require __DIR__ . '/../config/config.php';
switch ($config['publish']['type']) {
case 'database':
return new DatabasePublisher();
case 'file':
return new FilePublisher();
case 'api':
return new ApiPublisher();
default:
throw new Exception("Unknown publisher type");
}
}
private function getCollectionParams($type) {
$params = [
'url' => 'http://example.com/data',
'selector' => '.content'
];
// 这里可以读取配置或数据库中的任务参数
return $params;
}
}
// 命令行入口
if (php_sapi_name() === 'cli') {
$collector = new RunCollector();
$type = $argv[1] ?? 'html';
echo "Collection completed, published: " . $collector->execute($type) . " items\n";
}
Web管理界面
<?php
// admin/index.php - 管理系统入口
session_start();
// 检查登录状态
if (!isset($_SESSION['admin_logged_in'])) {
header('Location: login.php');
exit;
}
?>
<!DOCTYPE html>
<html>
<head>数据采集管理系统</title>
<style>
body { font-family: Arial, sans-serif; margin: 20px; }
.container { max-width: 1200px; margin: 0 auto; }
.header { background: #4CAF50; color: white; padding: 10px; }
.stat-box { display: inline-block; margin: 10px; padding: 20px; background: #f0f0f0; }
.btn { padding: 10px 20px; margin: 5px; cursor: pointer; }
table { border-collapse: collapse; width: 100%; margin-top: 20px; }
th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
th { background-color: #4CAF50; color: white; }
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>数据采集管理系统</h1>
<p>欢迎,<?php echo htmlspecialchars($_SESSION['username'] ?? '管理员'); ?> |
<a href="logout.php">退出登录</a></p>
</div>
<div class="stats">
<div class="stat-box">
<h3>今日采集量</h3>
<p><?php echo getTodayCount(); ?></p>
</div>
<div class="stat-box">
<h3>总数据量</h3>
<p><?php echo getTotalCount(); ?></p>
</div>
</div>
<div>
<h2>开始新采集任务</h2>
<form action="start_collection.php" method="post">
<select name="collector_type">
<option value="html">HTML页面</option>
<option value="api">API接口</option>
<option value="rss">RSS订阅</option>
</select>
<input type="url" name="url" placeholder="采集目标URL" required>
<input type="text" name="selector" placeholder="选择器" value="body">
<button type="submit" class="btn">开始采集</button>
</form>
</div>
<div>
<h2>最近采集数据</h2>
<table>
<tr>
<th>ID</th>
<th>标题</th>
<th>来源</th>
<th>采集时间</th>
<th>操作</th>
</tr>
<?php foreach (getLatestContent() as $content): ?>
<tr>
<td><?php echo $content['id']; ?></td>
<td><a href="view_content.php?id=<?php echo $content['id']; ?>">
<?php echo htmlspecialchars($content['title']); ?>
</a></td>
<td><a href="<?php echo htmlspecialchars($content['source_url']); ?>" target="_blank">查看原站</a></td>
<td><?php echo $content['created_at']; ?></td>
<td>
<a href="view_content.php?id=<?php echo $content['id']; ?>">查看</a> |
<a href="delete_content.php?id=<?php echo $content['id']; ?>"
onclick="return confirm('确定删除?')">删除</a>
</td>
</tr>
<?php endforeach; ?>
</table>
</div>
</div>
</body>
</html>
<?php
function getTodayCount() {
$db = Database::getInstance()->getConnection();
$stmt = $db->query("SELECT COUNT(*) FROM collected_data WHERE DATE(created_at) = CURDATE()");
return $stmt->fetchColumn();
}
function getTotalCount() {
$db = Database::getInstance()->getConnection();
$stmt = $db->query("SELECT COUNT(*) FROM collected_data");
return $stmt->fetchColumn();
}
function getLatestContent($limit = 20) {
$db = Database::getInstance()->getConnection();
$stmt = $db->query("SELECT * FROM collected_data ORDER BY created_at DESC LIMIT $limit");
return $stmt->fetchAll();
}
?>
使用说明
安装步骤:
- 创建数据库:
CREATE DATABASE collection_system; - 导入数据库结构(自动创建)
- 修改
config/config.php中的数据库配置 - 设置定时任务:
# crontab 配置 */5 * * * * php /path/to/cron/run_collector.php html */10 * * * * php /path/to/cron/run_collector.php api
使用方法:
# 命令行运行 php cron/run_collector.php html php cron/run_collector.php api php cron/run_collector.php rss
系统特点
- 模块化设计:采集器、解析器、发布器独立分离
- 多源支持:HTML、API、RSS等多种数据源
- 数据处理:自动去重、内容过滤、数据验证
- 灵活发布:支持数据库、文件、API等多种输出
- 完整日志:详细的操作日志记录
- Web管理:提供友好的管理界面
这个系统可以根据实际需求进行扩展,例如添加更多采集规则、数据清洗功能、实时监控等。