PHP 采集数据发布系统

wen PHP项目 2026-08-10 209

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

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管理：提供友好的管理界面

这个系统可以根据实际需求进行扩展,例如添加更多采集规则、数据清洗功能、实时监控等。

本文地址： https://www.bmjup.com/post/42954.html